/*****************************************************************************
 *  ENTROPY - emerging network to reduce orwellian potency yield
 *
 *  Copyright (C) 2002 Juergen Buchmueller <pullmoll@stop1984.com>
 *
 *  This program is free software; you can redistribute it and/or modify
 *  it under the terms of the GNU General Public License as published by
 *  the Free Software Foundation; either version 2 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU General Public License for more details.
 *
 *  You should have received a copy of the GNU General Public License
 *  along with this program; if not, write to the Free Software Foundation,
 *  Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA
 *
 *	$Id: peer.c,v 1.106 2005/12/04 20:35:45 pullmoll Exp $
 *****************************************************************************/
#include "peer.h"

/**
 * @brief Post ADV messages only to queues with fewer entries than this value
 */
#define	ADV_QUEUELIMIT (MSG_QUEUESIZE*3/8)

/**
 * @brief Post REQ messages only to queues with fewer entries than this value
 */
#define	REQ_QUEUELIMIT (MSG_QUEUESIZE*3/8)

/**
 * @brief Post INS messages only to queues with fewer entries than this value
 */
#define	INS_QUEUELIMIT (MSG_QUEUESIZE*5/8)

/**
 * @brief Post FEC messages only to queues with fewer entries than this value
 */
#define	FEC_QUEUELIMIT (MSG_QUEUESIZE*2/8)

/**
 * @brief Penalty delay types
 */
enum {
	PENALTY_SERVED,		/* message was served */
	PENALTY_IGNORE,		/* message was ignored */
	PENALTY_NFORWARD,	/* could not forward message */
	PENALTY_NHANDLE		/* could not handle message */
};

peer_t *g_peer = NULL;
pid_t g_peer_threads_pid = (pid_t)-1;
pid_t g_peer_bwlimit_pid = (pid_t)-1;
pid_t g_peer_fec_worker_pid = (pid_t)-1;
pid_t g_peer_out_pid = (pid_t)-1;
pid_t g_peer_in_pid = (pid_t)-1;

/**
 * @brief Acquire the peer global lock
 */
static inline int GLOBAL_LOCK(void)
{
	int rc;
	FUN("GLOBAL_LOCK");
	if (0 == g_peer->sem_global.magic) {
		errno = EINVAL;
		return -1;
	}
	rc = osd_sem_wait(&g_peer->sem_global);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("global lock failed (%s)\n",
			strerror(errno)));
		return rc;
	}
	return 0;
}

/**
 * @brief Release the peer global lock
 */
static inline int GLOBAL_UNLOCK(void)
{
	if (0 == g_peer->sem_global.magic) {
		errno = EINVAL;
		return -1;
	}
	return osd_sem_post(&g_peer->sem_global);
}

/**
 * @brief Acquire the peer bandwidth limiter lock
 */
static inline int BWLIMIT_LOCK(void)
{
	int rc;
	FUN("BWLIMIT_LOCK");
	if (0 == g_peer->sem_bwlimit.magic) {
		errno = EINVAL;
		return -1;
	}
	rc = osd_sem_wait(&g_peer->sem_bwlimit);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("bwlimit lock failed (%s)\n",
			strerror(errno)));
		return rc;
	}
	return 0;
}

/**
 * @brief Release the peer bandwidth limiter lock
 */
static inline int BWLIMIT_UNLOCK(void)
{
	if (0 == g_peer->sem_bwlimit.magic) {
		errno = EINVAL;
		return -1;
	}
	return osd_sem_post(&g_peer->sem_bwlimit);
}

/**
 * @brief Acquire the peer statistics lock
 */
static inline int STATS_LOCK(void)
{
	int rc;
	FUN("STATS_LOCK");
	if (0 == g_peer->sem_stats.magic) {
		errno = EINVAL;
		return -1;
	}
	rc = osd_sem_wait(&g_peer->sem_stats);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("stats lock failed (%s)\n",
			strerror(errno)));
		return rc;
	}
	return 0;
}

/**
 * @brief Release the peer statistics lock
 */
static inline int STATS_UNLOCK(void)
{
	if (0 == g_peer->sem_stats.magic) {
		errno = EINVAL;
		return -1;
	}
	return osd_sem_post(&g_peer->sem_stats);
}

/**
 * @brief Acquire the peer FEC message lock
 */
static inline int FEC_LOCK(void)
{
	int rc;
	FUN("FEC_LOCK");
	if (0 == g_peer->sem_fec.magic) {
		errno = EINVAL;
		return -1;
	}
	rc = osd_sem_wait(&g_peer->sem_fec);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("fec lock failed (%s)\n",
			strerror(errno)));
		return rc;
	}
	return 0;
}

/**
 * @brief Release the peer FEC message lock
 */
static inline int FEC_UNLOCK(void)
{
	if (0 == g_peer->sem_fec.magic) {
		errno = EINVAL;
		return -1;
	}
	return osd_sem_post(&g_peer->sem_fec);
}

/**
 * @brief Acquire the peer BCM message lock
 */
static inline int BCM_LOCK(void)
{
	int rc;
	FUN("BCM_LOCK");
	if (0 == g_peer->sem_bcm.magic) {
		errno = EINVAL;
		return -1;
	}
	rc = osd_sem_wait(&g_peer->sem_bcm);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("fec lock failed (%s)\n",
			strerror(errno)));
		return rc;
	}
	return 0;
}

/**
 * @brief Release the peer BCM message lock
 */
static inline int BCM_UNLOCK(void)
{
	if (0 == g_peer->sem_bcm.magic) {
		errno = EINVAL;
		return -1;
	}
	return osd_sem_post(&g_peer->sem_bcm);
}

/**
 * @brief Returns non zero if queue q is full
 */
#define	QUEUE_FULL(q) \
	(0 != q->size ? (((q->head + 1) % q->size) == q->tail) : 1)

/**
 * @brief Return number of queue slots left
 */
#define	QUEUE_SIZE(q) \
	((q->head >= q->tail) ? q->head - q->tail : q->size + q->head - q->tail)

/**
 * @brief Next entry for queue head, wrap at q->size
 */
#define	QUEUE_NEXT(q) \
	if (q->size == (q->head += 1)) q->head = 0

#undef	MIN
#undef	MAX
#define	MIN(a,b) ((a)<(b)?(a):(b))
#define	MAX(a,b) ((a)>(b)?(a):(b))

#define	EK5_ENTROPY	"ENTROPY"

static void peer_out_exit(int sig);
static void peer_in_exit(int sig);
static void peer_bwlimit_exit(int sig);
static void peer_fec_worker_exit(int sig);
static void peer_exit(int sig);
static int peer_seednodes(int *found);
static int peer_adv_cancel(sha1_digest_t *s, struct sockaddr_in *peeraddr);
static int peer_req_cancel(sha1_digest_t *s);
static int peer_ins_cancel(sha1_digest_t *s, struct sockaddr_in *peeraddr);
static int peer_blacklist_check(struct sockaddr_in *peeraddr);
#if	DEBUG
/**
 * @brief Return debug name for a message
 */
static const char *msg_name(int msg)
{
	switch (msg) {
	case MSG_CAN: return "CAN";
	case MSG_ADV: return "ADV";
	case MSG_REQ: return "REQ";
	case MSG_INS: return "INS";
	case MSG_ANN: return "ANN";
	case MSG_FPR: return "FPR";
	case MSG_FEC: return "FEC";
	case MSG_BCM: return "BCM";
	}
	return "???";
}
#endif


/**
 * @brief loglevel for penalty_delay()
 */
#define	L_PENALTY L_DEBUG

/**
 * @brief Penalty delay after a message was received
 *
 * While working down incoming messages, the node detected a reason to
 * impose a penalty delay after the message. Depending on message type
 * and reason, why the message is worth a penalty, the delay is
 * built from the distance of the key's routing from our fingerprint
 * and a fixed factor.
 *
 * @param fpr fingerprint slot of the message
 * @param msg message type
 * @param msg reason why a penalty is due
 */
static void penalty_delay(size_t fpr, int msg, int reason)
{
	int64_t delay = 0;
	FUN("penalty_delay");

	fpr %= 16;
	/* delay is distance from fingerprint peaks of our node */
	fpr = 255 - g_store->fingerprint[fpr];
	switch (msg) {
	case MSG_ADV:	/* key advertizements */
		switch (reason) {
		case PENALTY_SERVED:
			/* we served the advertizement by requesting */
			delay = fpr * 500;
			/* NOLOG */
			break;
		case PENALTY_IGNORE:
			/* we ignored the advertizement */
			delay = fpr * 1000;
			LOGS(L_PEER,L_PENALTY,("ADV %02x ignore: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NFORWARD:
			/* we can't forward the advertizement */
			delay = fpr * 1000;
			LOGS(L_PEER,L_PENALTY,("ADV %02x can't fwd: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NHANDLE:
			/* we can't request the key from the advertizer */
			delay = fpr * 5000;
			LOGS(L_PEER,L_PENALTY,("ADV %02x can't req: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		}
		break;
	case MSG_REQ:	/* key requests */
		switch (reason) {
		case PENALTY_SERVED:
			/* we served the request by inserting */
			delay = fpr * 500;
			/* NOLOG */
			break;
		case PENALTY_IGNORE:
			/* we ignored the request */
			delay = fpr * 5000;
			LOGS(L_PEER,L_PENALTY,("REQ %02x ignore: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NFORWARD:
			/* we can't forward the request */
			delay = fpr * 1000;
			LOGS(L_PEER,L_PENALTY,("REQ %02x can't fwd: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NHANDLE:
			/* we can't insert the key to the requester */
			delay = fpr * 10000;
			LOGS(L_PEER,L_PENALTY,("REQ %02x can't ins: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		}
		break;
	case MSG_INS:
		switch (reason) {
		case PENALTY_SERVED:
			/* we served the insert by advertizing */
			delay = fpr * 500;
			/* NOLOG */
			break;
		case PENALTY_IGNORE:
			/* we ignored the insert */
			delay = fpr * 10000;
			LOGS(L_PEER,L_PENALTY,("INS %02x ignore: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NFORWARD:
			/* we can't forward an advertizement for this key */
			delay = fpr * 1000;
			LOGS(L_PEER,L_PENALTY,("INS %02x can't fwd: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		case PENALTY_NHANDLE:
			/* we can't request a bad key from someone else */
			delay = fpr * 1000;
			LOGS(L_PEER,L_PENALTY,("INS %02x can't req: %s\n",
				(unsigned)fpr, usec_str(delay)));
			break;
		}
		break;
	}
	osd_usleep(delay);
}


/**
 * @brief Return a fingerprint threshold for try with HTL
 *
 * Determine a fingerprint threshold for a given try number
 * and HTL (hops to live) value. If fewer than PEERS_OUT_MAX
 * nodes are connected, make the threshold 'wider' than usual.
 *
 * For HTL <= 0 always return the maximum.
 * For HTL > 0 scale to the range 255 downto 0 for 1 to maxhtl/try
 * For HTL >= maxhtl/width return 0
 *
 * @param try The number of the retry (1...g_conf->fcpretries)
 * @param htl The hops-to-live value
 *
 * @result A threshold level between 0 and 255
 */
static uint32_t threshold_for_htl(int try, int htl)
{
	int cutoff;
	int width;

	if (htl < 0)
		return 255;
	if (htl > g_conf->maxhtl)
		htl = g_conf->maxhtl;
	if (0 == g_peer->peercnt_out)
		return 255;
	if (try < 1)
		try = 1;
	width = PEERS_OUT_MAX / g_peer->peercnt_out;
	cutoff = try * width * 255 * htl / g_conf->maxhtl;
	if (cutoff > 255)
		cutoff = 255;
	return 255 - cutoff;
}

/**
 * @brief Return 0 if peer passes blacklist check
 *
 * Check whether the peer address is blacklisted
 * and return non-0 if the address is in the blacklist
 * and not explicitly whitelisted.  Return 0 if ok.
 *
 * @param peeraddr The address of the peer
 *
 * @result Return 0 if peer is ok
 */

static int peer_blacklist_check(struct sockaddr_in *peeraddr)
{
	char *bldescr = NULL, *whdescr = NULL;
	FUN("peer_blacklist_check");

	if (ipv4_black_list_match(peeraddr->sin_addr.s_addr, &bldescr)) {
		if (ipv4_white_list_match(peeraddr->sin_addr.s_addr, &whdescr)) {
			LOGS(L_PEER,L_NORMAL,
				 ("Peer %s is blacklisted (%s) "
				  "but also whitelisted (%s).\n",
				  sock_ntoa(peeraddr),
				  (bldescr != NULL) ? bldescr : "",
				  (whdescr != NULL) ? whdescr : ""));
		 } else {
			 LOGS(L_PEER,L_NORMAL,
				  ("Peer %s is blacklisted (%s).\n",
				   sock_ntoa(peeraddr),
				   (bldescr != NULL) ? bldescr : ""));
			 return -1;
		 }
	} else {
		LOGS(L_PEER,L_MINOR,
			 ("Peer %s is acceptable.\n", sock_ntoa(peeraddr)));
	}
	return 0;
}

/**
 * @brief Return a spread factor for a given HTL
 *
 * Determine a spread factor (# of nodes) for a given HTL value.
 * HTL range is mapped to 2..peercnt_out/2
 *
 * @param htl The hops-to-live value
 *
 * @result Return a (max.) number of peers to spread a key to
 */
static uint32_t spread_for_htl(int htl)
{
	uint32_t spr;

	if (htl <= 0)
		return 0;
	if (htl > g_conf->maxhtl)
		htl = g_conf->maxhtl;
	if (g_peer->peercnt_out >= 4) {
		spr = (g_peer->peercnt_out / 2 - 2) * htl / g_conf->maxhtl + 2;
	} else {
		spr = g_peer->peercnt_out / 2;
	}
	if (spr > g_peer->peercnt_out / 2)
		spr = g_peer->peercnt_out;
	return spr;
}


/**
 * @brief Maximum size of the initial='' attribute of the crypto tag in bytes
 */
#define	MAXINITIAL	16384

/**
 * @brief structure to keep the parsed XML p2p info
 *
 * Format of the p2p_info XML texts circulating the network:
 * <pre>
 *	<?xml version="1.0" standalone='yes'?>
 *	<p2p>
 *		<peer address='something.dyndns.org:37373' />
 *		<node type='entropy' major='0' minor='3.0' build='176' />
 *		<store fingerprint='11...[32 digits]...00' size='xxx' current='xxx' />
 *		<crypto module='crypt2' size='xxx' encoded='xxx' initial='1234....' />
 *	</p2p>
 * </pre>
 *
 */
typedef struct p2p_info_s {
	struct {
		int found_tag;
		struct sockaddr_in addr;
		char hostname[MAXPATHLEN];
	}	peer;
	struct {
		int found_tag;
		char type[32+1];
		uint32_t major;
		uint32_t minor;
		uint32_t build;
	}	node;
	struct {
		int found_tag;
		uint8_t fingerprint[16];
		uint32_t fingerprint_sum;
		char fingerprint_str[2*16+1];
		uint64_t size;
		uint64_t current;
	}	store;
	struct {
		int found_tag;
		char module[32+1];
		size_t size;
		size_t encoded;
		uint8_t initial[MAXINITIAL];
		char initial_str[2*MAXINITIAL+1];
	}	crypto;
}	p2p_info_t;

/**
 * @brief structure to keep the parsed XML broadcast info
 *
 * Format of the broadcast XML texts circulating the network:
 * <pre>
 *	<?xml version="1.0" standalone='yes'?>
 *	<broadcast>
 *		<channel name="channel name" />
 *		<sender name="sender name" />
 *		<payload text="some text" />
 *	</broadcast>
 * </pre>
 *
 */
typedef struct broadcast_s {
	struct {
		int found_tag;
		char name[MAXCHANNEL];
	}	channel;
	struct {
		int found_tag;
		char name[MAXSENDER];
	}	sender;
	struct {
		int found_tag;
		char text[MAXPAYLOAD];
	}	payload;
}	broadcast_t;

static int peer_get_name(struct sockaddr_in *addr,
	char **name);
static int peer_set_name(struct sockaddr_in *addr,
	const char *peername);
static int peer_set_fingerprint(struct sockaddr_in *addr,
	uint8_t *fingerprint);
static int peer_set_crypto_module_in(struct sockaddr_in *addr,
	size_t crypto_module);
static int peer_set_size_current(struct sockaddr_in *addr,
	uint64_t storesize, uint64_t currsize);
static int peer_set_node(struct sockaddr_in *addr,
	const char *type, uint32_t major, uint32_t minor, uint32_t build);

static int peer_set_cancer(struct sockaddr_in *addr, int cancer_node);

#if	DEBUG
/**
 * @brief Convert control characters to readable mnemonics for debug output
 */
static const char *ascii_dump(const char *src, int len)
{
	static char buff[8][128];
	static int which = 0;
	char *dst;

	which = (which + 1) % 8;
	dst = buff[which];
	for (dst = buff[which]; len-- > 0 && dst < &buff[which][120]; /* */) {
		switch (*src) {
		case '\b': dst += sprintf(dst, "[BS]"); break;
		case '\t': dst += sprintf(dst, "[TAB]"); break;
		case '\r': dst += sprintf(dst, "[LF]"); break;
		case '\n': dst += sprintf(dst, "[CR]"); break;
		default: *dst += sprintf(dst, "%c", *src);
		}
		src++;
	}
	*dst = '\0';

	return buff[which];
}
#endif

/**
 * @brief Escape a source string which might contain XML special characters
 *
 * Replace every occurence of < > & and ' by their respective
 * entity encodings: &lt; &gt; &amp; and &#39;
 *
 * @param src pointer to the source string
 *
 * @result returns an escaped copy
 */
static const char *xml_escape(const char *src)
{
	static char *buff[8];
	static int which = 0;
	char *dst, *end;
	size_t len;

	which = (which + 1) % 8;
	xfree(buff[which]);
	len = 5 * strlen(src) + 1;
	dst = buff[which] = xcalloc(len, sizeof(char));
	end = &dst[len - 1];
	
	while (*src) {
		if (*src == '<')
			dst += pm_snprintf(dst, (int)(end - dst), "&lt;");
		else if (*src == '>')
			dst += pm_snprintf(dst, (int)(end - dst), "&gt;");
		else if (*src == '&')
			dst += pm_snprintf(dst, (int)(end - dst), "&amp;");
		else if (*src == 39)
			dst += pm_snprintf(dst, (int)(end - dst), "&#39;");
		else
			*dst++ = *src;
		src++;
		if (dst == end)
			break;
	}
	*dst = '\0';
	return buff[which];
}

/**
 * @brief Callback for the XML 'peer' tag start
 */
static void p2p_info_xml_start_peer(void *user, const char **atts)
{
	struct sockaddr_in addr;
	p2p_info_t *pi = (p2p_info_t *)user;
	const char **att;
	const char *sol;
	char *eol;
	int len;
	FUN("p2p_info_xml_start_peer");

	memset(&pi->peer, 0, sizeof(pi->peer));
	pi->peer.found_tag = 1;

	memset(&addr, 0, sizeof(addr));
	/* Default to AF_INET (should we ever support AF_INET6 etc.) */
	addr.sin_family = AF_INET;
	/* Default to invalid address 255.255.255.255 if no/wrong hostname */
	addr.sin_addr.s_addr = (uint32_t)-1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "hostname")) {
			sol = att[1];
			/* strip leading blanks */
			while (*sol && isspace(*sol))
				sol++;
			len = pm_snprintf(pi->peer.hostname, sizeof(pi->peer.hostname),
				"%s", sol);
			/* strip trailing blanks */
			eol = pi->peer.hostname + len - 1;
			while (eol >= pi->peer.hostname && isspace(*eol))
				eol--;
			eol++;
			*eol = '\0';
			LOGS(L_PEER,L_MINOR,("hostname: '%s'\n",
				pi->peer.hostname));
		} else if (0 == strcasecmp(att[0], "port")) {
			LOGS(L_PEER,L_MINOR,("port: %s\n", att[1]));
			addr.sin_port = ntohs(strtoul(att[1], NULL, 0));
		} else if (0 == strcasecmp(att[0], "address")) {
			LOGS(L_PEER,L_MINOR,("address: %s\n", att[1]));
			if (0 != sock_aton(&addr, att[1])) {
				LOGS(L_PEER,L_ERROR,("invalid address: %s (%s)\n",
					att[1], strerror(errno)));
			}
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <peer> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
	/* set the address which came from hostname:port or address tags */
	pi->peer.addr = addr;
}

/**
 * @brief Callback for the XML 'node' tag start
 */
static void p2p_info_xml_start_node(void *user, const char **atts)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	const char **att;
	FUN("p2p_info_xml_start_node");

	memset(&pi->node, 0, sizeof(pi->node));
	pi->node.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "type")) {
			LOGS(L_PEER,L_DEBUG,("type: %s\n", att[1]));
			strncpy(pi->node.type, att[1],
				sizeof(pi->node.type));
		} else if (0 == strcasecmp(att[0], "major")) {
			LOGS(L_PEER,L_DEBUG,("major: %s\n", att[1]));
			pi->node.major = strtoul(att[1], NULL, 0);
		} else if (0 == strcasecmp(att[0], "minor")) {
			char minor[32+1], *dot;
			LOGS(L_PEER,L_DEBUG,("minor: %s\n", att[1]));
			strncpy(minor, att[1], sizeof(minor));
			dot = strchr(minor, '.');
			if (NULL == dot) {
				pi->node.minor = strtoul(minor, NULL, 0) << 16;
			} else {
				*dot++ = '\0';
				pi->node.minor = (strtoul(minor, NULL, 0) << 16) +
					(strtoul(dot, NULL, 0) % 0x10000);
			}
		} else if (0 == strcasecmp(att[0], "build")) {
			LOGS(L_PEER,L_DEBUG,("build: %s\n", att[1]));
			pi->node.build = strtoul(att[1], NULL, 0);
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <node ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for the XML 'store' tag start
 */
static void p2p_info_xml_start_store(void *user, const char **atts)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	const char **att;
	FUN("p2p_info_xml_start_store");

	memset(&pi->store, 0, sizeof(pi->store));
	pi->store.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "fingerprint")) {
			LOGS(L_PEER,L_DEBUG,("fingerprint: %s\n", att[1]));
			strncpy(pi->store.fingerprint_str, att[1],
				sizeof(pi->store.fingerprint_str));
		} else if (0 == strcasecmp(att[0], "size")) {
			LOGS(L_PEER,L_DEBUG,("size: %s\n", att[1]));
			pi->store.size = strtoull(att[1], NULL, 16);
		} else if (0 == strcasecmp(att[0], "current")) {
			LOGS(L_PEER,L_DEBUG,("current: %s\n", att[1]));
			pi->store.current = strtoull(att[1], NULL, 16);
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <store ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for the XML 'crypto' tag start
 */
static void p2p_info_xml_start_crypto(void *user, const char **atts)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	const char **att;
	FUN("p2p_info_xml_start_crypto");

	memset(&pi->crypto, 0, sizeof(pi->crypto));
	pi->crypto.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "module")) {
			LOGS(L_PEER,L_DEBUG,("module: %s\n", att[1]));
			pm_snprintf(pi->crypto.module, sizeof(pi->crypto.module),
				"%s", att[1]);
		} else if (0 == strcasecmp(att[0], "size")) {
			LOGS(L_PEER,L_DEBUG,("size: %s\n", att[1]));
			pi->crypto.size = strtoul(att[1], NULL, 16);
		} else if (0 == strcasecmp(att[0], "encoded")) {
			LOGS(L_PEER,L_DEBUG,("encoded: %s\n", att[1]));
			pi->crypto.encoded = strtoul(att[1], NULL, 16);
			if (pi->crypto.encoded > MAXINITIAL) {
				pi->crypto.encoded = MAXINITIAL;
				LOGS(L_PEER,L_ERROR,("clipped crypto.encoded to %d (this will fail).\n",
					MAXINITIAL));
			}
		} else if (0 == strcasecmp(att[0], "initial")) {
			LOGS(L_PEER,L_DEBUG,("initial: %s\n", att[1]));
			pm_snprintf(pi->crypto.initial_str, sizeof(pi->crypto.initial_str),
				"%s", att[1]);
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <crypto ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for any XML 'p2p' tag start
 */
static void p2p_info_xml_start_element(void *user, const char *name,
				const char **atts)
{
	FUN("p2p_info_xml_start_element");

	LOGS(L_PEER,L_DEBUG,("start tag: %s\n", name));
	if (0 == strcasecmp(name, "p2p")) {
		/* following are the tags */
	} else if (0 == strcasecmp(name, "peer")) {
		p2p_info_xml_start_peer(user, atts);
	} else if (0 == strcasecmp(name, "node")) {
		p2p_info_xml_start_node(user, atts);
	} else if (0 == strcasecmp(name, "store")) {
		p2p_info_xml_start_store(user, atts);
	} else if (0 == strcasecmp(name, "crypto")) {
		p2p_info_xml_start_crypto(user, atts);
	} else {
		LOGS(L_PEER,L_ERROR,("ignoring unknown start tag: %s\n", name));
	}
}

#define	XDIGIT(n) (((n) > '9' ? ((n)-'A'+10) : (n)) % 16)
/**
 * @brief Callback for any XML 'p2p' tag end
 */
static void p2p_info_xml_end_element(void *user, const char *name)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	FUN("p2p_info_xml_end_element");

	LOGS(L_PEER,L_DEBUG,("end tag: %s\n", name));
	if (0 == strcasecmp(name, "p2p")) {
		/* currently nothing's left to do here */
	} else if (0 == strcasecmp(name, "peer")) {
		/* currently nothing's left to do here */
	} else if (0 == strcasecmp(name, "node")) {
		/* currently nothing's left to do here */
	} else if (0 == strcasecmp(name, "store")) {
		const char *src = pi->store.fingerprint_str;
		uint8_t *dst = pi->store.fingerprint;
		size_t i;

		/* convert the hex 'fingerprint' to binary */
		pi->store.fingerprint_sum = 0;
		for (i = 0; i < 16; i++, src += 2) {
			dst[i] = 16 * XDIGIT(src[0]) + XDIGIT(src[1]);
			pi->store.fingerprint_sum += dst[i];
		}
	} else if (0 == strcasecmp(name, "crypto")) {
		if (pi->crypto.encoded > 0) {
			const char *src = pi->crypto.initial_str;
			uint8_t *dst = pi->crypto.initial;
			size_t i;

			/* convert the hex 'initial' to binary */
			for (i = 0; i < pi->crypto.encoded; i++, src += 2) {
				dst[i] = 16 * XDIGIT(src[0]) + XDIGIT(src[1]);
			}
		}
	} else {
		LOGS(L_PEER,L_ERROR,("ignoring unknown end tag: %s\n", name));
	}
}

/**
 * @brief Callback for XML CDATA
 */
static void p2p_info_xml_cdata(void *user, const char *s, int len)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	FUN("p2p_info_xml_cdata");
	/* hmmm... */
	(void)pi;
	(void)s;
	(void)len;
	LOGS(L_PEER,L_DEBUGX,("cdata (%d): %s\n", len, ascii_dump(s, len)));
}

/**
 * @brief Callback for XML junk (blanks, tabs, newlines, etc.)
 */
static void p2p_info_xml_default(void *user, const char *s, int len)
{
	p2p_info_t *pi = (p2p_info_t *)user;
	FUN("p2p_info_xml_default");
	/* hmmm... */
	(void)pi;
	(void)s;
	(void)len;
	LOGS(L_PEER,L_DEBUGX,("default (%d): %s\n", len, ascii_dump(s, len)));
}

/**
 * @brief Callback for the XML 'channel' tag start
 */
static void broadcast_xml_start_channel(void *user, const char **atts)
{
	broadcast_t *bc = (broadcast_t *)user;
	const char **att;
	FUN("broadcast_xml_start_channel");

	memset(&bc->channel, 0, sizeof(bc->channel));
	bc->channel.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "name")) {
			strncpy(bc->channel.name, att[1],
				sizeof(bc->channel.name));
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <channel ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for the XML 'sender' tag start
 */
static void broadcast_xml_start_sender(void *user, const char **atts)
{
	broadcast_t *bc = (broadcast_t *)user;
	const char **att;
	FUN("broadcast_xml_start_sender");

	memset(&bc->sender, 0, sizeof(bc->sender));
	bc->sender.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "name")) {
			strncpy(bc->sender.name, att[1],
				sizeof(bc->sender.name));
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <sender ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for the XML 'payload' tag start
 */
static void broadcast_xml_start_payload(void *user, const char **atts)
{
	broadcast_t *bc = (broadcast_t *)user;
	const char **att;
	FUN("broadcast_xml_start_payload");

	memset(&bc->payload, 0, sizeof(bc->payload));
	bc->payload.found_tag = 1;
	for (att = atts; *att; att += 2) {
		if (0 == strcasecmp(att[0], "text")) {
			strncpy(bc->payload.text, att[1],
				sizeof(bc->payload.text));
		} else {
			LOGS(L_PEER,L_ERROR,("unhandled <payload ...> attribute: %s='%s'\n",
				att[0], att[1]));
		}
	}
}

/**
 * @brief Callback for any XML 'broadcast' tag start
 */
static void broadcast_xml_start_element(void *user, const char *name,
				const char **atts)
{
	FUN("broadcast_xml_start_element");

	LOGS(L_PEER,L_DEBUG,("start tag: %s\n", name));
	if (0 == strcasecmp(name, "broadcast")) {
		/* following are the tags */
	} else if (0 == strcasecmp(name, "channel")) {
		broadcast_xml_start_channel(user, atts);
	} else if (0 == strcasecmp(name, "sender")) {
		broadcast_xml_start_sender(user, atts);
	} else if (0 == strcasecmp(name, "payload")) {
		broadcast_xml_start_payload(user, atts);
	} else {
		LOGS(L_PEER,L_ERROR,("ignoring unknown start tag: %s\n", name));
	}
}

/**
 * @brief Callback for any XML 'broadcast' tag end
 */
static void broadcast_xml_end_element(void *user, const char *name)
{
	broadcast_t *bc = (broadcast_t *)user;
	FUN("broadcast_xml_end_element");

	LOGS(L_PEER,L_DEBUG,("end tag: %s\n", name));
	if (0 == strcasecmp(name, "broadcast")) {
		/* currently nothing's left to do here */
	} else if (0 == strcasecmp(name, "channel")) {
		bc->channel.name[sizeof(bc->channel.name) - 1] = '\0';
	} else if (0 == strcasecmp(name, "sender")) {
		bc->sender.name[sizeof(bc->sender.name) - 1] = '\0';
	} else if (0 == strcasecmp(name, "payload")) {
		bc->payload.text[sizeof(bc->payload.text) - 1] = '\0';
	} else {
		LOGS(L_PEER,L_ERROR,("ignoring unknown end tag: %s\n", name));
	}
}

/**
 * @brief Callback for any XML 'broadcast' CDATA
 */
static void broadcast_xml_cdata(void *user, const char *s, int len)
{
	broadcast_t *bc = (broadcast_t *)user;
	FUN("broadcast_xml_cdata");
	/* hmmm... */
	(void)bc;
	(void)s;
	(void)len;
	LOGS(L_PEER,L_DEBUGX,("cdata (%d): %s\n", len, ascii_dump(s, len)));
}

/**
 * @brief Callback for any XML 'broadcast' junk (blanks, tabs, newlines, etc.)
 */
static void broadcast_xml_default(void *user, const char *s, int len)
{
	broadcast_t *bc = (broadcast_t *)user;
	FUN("broadcast_xml_default");
	/* hmmm... */
	(void)bc;
	(void)s;
	(void)len;
	LOGS(L_PEER,L_DEBUGX,("default (%d): %s\n", len, ascii_dump(s, len)));
}

/**
 * @brief Make a new entry into the list of inbound peer nodes.
 *
 * Get a piece of shared memory for a peer_node_t with
 * a msg_queue_t of size 0 (just the size).
 * Check if another inbound connection is acceptable and
 * make a new entry into the list of inbound peer nodes.
 * Otherwise free the shared memory again and return NULL.
 *
 * @result NULL on error, else pointer to new peer_node_t
 */
peer_node_t *peer_add_in(void)
{
	peer_node_t *p = NULL, *n = NULL;
	size_t i, j, size;
	FUN("peer_add_in");

	/* for inbound connections don't allocate a msg_queue_t, only size_t */
	size = sizeof(peer_node_t) - sizeof(msg_queue_t) + sizeof(size_t);
	n = (peer_node_t *)scalloc(1, size);
	if (NULL == n) {
		LOGS(L_PEER,L_ERROR,("scalloc(1,%#x) failed (%s)\n",
			(unsigned)size, strerror(errno)));
		errno = ENOMEM;
		return NULL;
	}
	LOGS(L_PEER,L_MINOR,("new peer in shared memory: %#x at %p\n",
		(unsigned)size, n));

	if (0 != GLOBAL_LOCK()) {
		sfree((caddr_t)n);
		return NULL;
	}

	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			for (j = i + 1; j < PEERS_IN_MAX - 1; j++)
				g_peer->peers_in[j] = g_peer->peers_in[j+1];
			g_peer->peercnt_in -= 1;
			g_peer->peers_in[g_peer->peercnt_in] = NULL;
			LOGS(L_PEER,L_MINOR,("removed empty slot #%d\n", (int)i));
			i -= 1;
			continue;
		}
	}

	if ((i = g_peer->peercnt_in) < PEERS_IN_MAX) {
		p = n;
		p->pid = getpid();
		p->conn.socket = -1;
		p->queue.size = 0;
		g_peer->peers_in[i] = p;
		g_peer->peercnt_in = i + 1;
		GLOBAL_UNLOCK();
		LOGS(L_PEER,L_DEBUG,("create peer #%d PID {%d}\n",
			(int)i, (int)p->pid));
		return p;
	}

	GLOBAL_UNLOCK();
	sfree((caddr_t)n);
	errno = EAGAIN;
	return NULL;
}

/**
 * @brief Make a new entry into the list of outbound peer nodes.
 *
 * Get a piece of shared memory for a peer_node_t with
 * a msg_queue_t of size sizeof(msg_queue_t), i.e. _with_ queue.
 * Check if another outbound connection is possible and
 * make a new entry into the list of outbound peer nodes.
 * Otherwise free the shared memory again and return NULL.
 *
 * @result NULL on error, else pointer to new peer_node_t
 */
peer_node_t *peer_add_out(void)
{
	peer_node_t *p = NULL, *n = NULL;
	size_t i, j, size;
	FUN("peer_add_out");

	/* for outbound connections allocate a full peer_node_t */
	size = sizeof(peer_node_t);
	n = (peer_node_t *)scalloc(1, size);
	if (NULL == n) {
		LOGS(L_PEER,L_ERROR,("scalloc(1,%#x) failed (%s)\n",
			(unsigned)size, strerror(errno)));
		errno = ENOMEM;
		return NULL;
	}
	LOGS(L_PEER,L_MINOR,("new peer out shared memory: %#x at %p\n",
		(unsigned)size, n));

	if (0 != GLOBAL_LOCK()) {
		sfree((caddr_t)n);
		return NULL;
	}

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			for (j = i + 1; j < PEERS_OUT_MAX - 1; j++)
				g_peer->peers_out[j] = g_peer->peers_out[j+1];
			g_peer->peercnt_out -= 1;
			g_peer->peers_out[g_peer->peercnt_out] = NULL;
			LOGS(L_PEER,L_MINOR,("removed empty slot #%d\n", (int)i));
			i -= 1;
			continue;
		}
	}

	if ((i = g_peer->peercnt_out) < PEERS_OUT_MAX) {
		p = n;
		p->pid = getpid();
		p->conn.socket = -1;
		p->queue.size = MSG_QUEUESIZE;
		g_peer->peers_out[i] = p;
		g_peer->peercnt_out = i + 1;
		GLOBAL_UNLOCK();
		LOGS(L_PEER,L_DEBUG,("create peer #%d PID {%d}\n",
			(int)i, (int)p->pid));
		return p;
	}

	GLOBAL_UNLOCK();
	sfree((caddr_t)n);
	errno = EAGAIN;
	return NULL;
}

/**
 * @brief Remove an entry from the list of inbound peer nodes.
 *
 * Search for a slot containing the pointer to the shared
 * memory 'p'. If it is found, destroy the crypto module's
 * context, move down the list of peer nodes and finally
 * free the shared memory.
 * If 'p' is not found for some reason (BUG?), still free
 * the shared memory.
 *
 * @param p pointer to a peer_node_t
 *
 * @result zero on succes, -1 and errno = ENOENT if not found
 */
int peer_del_in(peer_node_t *p)
{
	size_t i, j;
	FUN("peer_del_in");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (p != g_peer->peers_in[i]) {
			/* wrong slot */
			continue;
		}
		LOGS(L_PEER,L_DEBUG,("found peer_node_t %p in slot #%u\n",
			p, (unsigned)i));
		crypto_exit(p->crypto_module_out, p->crypto_ctx);
		LOGS(L_PEER,L_DEBUG,("move down slots #%u to #%u\n",
			(unsigned)i, (unsigned)(g_peer->peercnt_in - 1)));
		for (j = i; j < g_peer->peercnt_in - 1; j++)
			g_peer->peers_in[j] = g_peer->peers_in[j+1];
		g_peer->peercnt_in -= 1;
		g_peer->peers_in[g_peer->peercnt_in] = NULL;
		LOGS(L_PEER,L_DEBUG,("peercnt_in is now #%u\n",
			(unsigned)g_peer->peercnt_in));
		GLOBAL_UNLOCK();
		sfree((caddr_t)p);
		LOGS(L_PEER,L_DEBUG,("done with peer_node_t %p\n",
			p));
		return 0;
	}
	GLOBAL_UNLOCK();
	LOGS(L_PEER,L_DEBUG,("peer_node_t %p not found\n",
		p));
	sfree((caddr_t)p);
	errno = ENOENT;
	return -1;
}

/**
 * @brief Remove an entry from the list of outbound peer nodes.
 *
 * Search for a slot containing the pointer to the shared
 * memory 'p'. If it is found, destroy the crypto module's
 * context, move down the list of peer nodes and finally
 * free the shared memory.
 * If 'p' is not found for some reason (BUG?), still free
 * the shared memory.
 *
 * @param p pointer to a peer_node_t
 *
 * @result zero on succes, -1 and errno = ENOENT if not found
 */
int peer_del_out(peer_node_t *p)
{
	size_t i, j;
	FUN("peer_del_out");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (p != g_peer->peers_out[i]) {
			/* wrong slot */
			continue;
		}
		LOGS(L_PEER,L_DEBUG,("found peer_node_t %p in slot #%u\n",
			p, (unsigned)i));
		crypto_exit(p->crypto_module_out, p->crypto_ctx);
		LOGS(L_PEER,L_DEBUG,("move down slots #%u to #%u\n",
			(unsigned)i, (unsigned)(g_peer->peercnt_out - 1)));
		for (j = i; j < g_peer->peercnt_out - 1; j++)
			g_peer->peers_out[j] = g_peer->peers_out[j+1];
		g_peer->peercnt_out -= 1;
		g_peer->peers_out[g_peer->peercnt_out] = NULL;
		LOGS(L_PEER,L_DEBUG,("peercnt_out is now #%u\n",
			(unsigned)g_peer->peercnt_out));
		GLOBAL_UNLOCK();
		sfree((caddr_t)p);
		LOGS(L_PEER,L_DEBUG,("done with peer_node_t %p\n",
			p));
		return 0;
	}
	GLOBAL_UNLOCK();
	LOGS(L_PEER,L_DEBUG,("peer_node_t %p not found\n",
		p));
	sfree((caddr_t)p);
	errno = ENOENT;
	return -1;
}

/**
 * @brief Compute the active inbound connection count
 *
 * Acquire the peer global lock, scan the list of inbound
 * connections with live sockets and connected status,
 * and count them. Release the peer global lock and
 * return the count.
 *
 * @result Number of active inbound connections
 */
int peer_in_count(void)
{
	peer_node_t *p;
	size_t i;
	int n = 0;
	FUN("peer_in_count");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		n++;
	}
	GLOBAL_UNLOCK();

	return n;
}

/**
 * @brief Compute the active outbound connection count
 *
 * Acquire the peer global lock, scan the list of outbound
 * connections with live sockets and connected status,
 * and count them. Release the peer global lock and
 * return the count.
 *
 * @result Number of active outbound connections
 */
int peer_out_count(void)
{
	peer_node_t *p;
	size_t i;
	int n = 0;
	FUN("peer_out_count");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		n++;
	}
	GLOBAL_UNLOCK();

	return n;
}

/**
 * @brief Check if we are already connected to a peer address
 *
 * Return zero if the peer specified in 'addr' is already connected,
 * otherwise return -1 and set errno to ENOENT.
 *
 * @param addr pointer to a struct sockaddr_in of the peer
 *
 * @result zero if found, -1 and errno = ENOENT if not found
 */
int peer_connected(struct sockaddr_in *addr)
{
	peer_node_t *p;
	size_t i;
	FUN("peer_connected");

	if (addr->sin_addr.s_addr == g_conf->node.sin_addr.s_addr &&
		addr->sin_port == g_conf->node.sin_port) {
		LOGS(L_PEER,L_DEBUGX,("found %s -- this is our node\n",
			sock_ntoa(addr)));
		return 0;	/* return if this is our own address */
	}

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CLOSED == p->conn.status) {
			/* connection closed */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr ||
			p->conn.address.sin_port != addr->sin_port) {
			/* not the node we want */
			continue;
		}
		GLOBAL_UNLOCK();
		LOGS(L_PEER,L_DEBUGX,("found %s -- connected since %s\n",
			sock_ntoa(addr), datetime_str(p->start)));
		return 0;
	}
	GLOBAL_UNLOCK();
	errno = ENOENT;
	return -1;
}

/**
 * @brief Check if we are trying to connect to a peer address
 *
 * Return zero if the peer specified address is in the pipe of
 * peers to try to connect, otherwise return -1 and set
 * errno to ENOENT.
 *
 * @param addr pointer to a struct sockaddr_in of the peer
 *
 * @result zero if found, -1 and errno = ENOENT if not found
 */
int peer_connecting(const struct sockaddr_in *addr)
{
	addr_time_t *c;
	time_t t;
	size_t i, j;
	FUN("peer_connecting");

	if (0 != GLOBAL_LOCK())
		return -1;

	t = time(NULL);

	for (i = 0; i < g_peer->connect_cnt; i++) {
		c = &g_peer->connect[i];
		if (c->time < t) {
			LOGS(L_PEER,L_MINOR,("connect entry #%d node %s is done (%s)\n",
				(int)i, sock_ntoa(&c->addr), datetime_str(c->time)));
			for (j = i; j < g_peer->connect_cnt - 1; j++)
				g_peer->connect[j] = g_peer->connect[j+1];
			g_peer->connect_cnt -= 1;
			i--;
			continue;
		}
		if (c->addr.sin_addr.s_addr == addr->sin_addr.s_addr &&
			c->addr.sin_port == addr->sin_port) {
			LOGS(L_PEER,L_MINOR,("won't try to connect node %s until %s\n",
				sock_ntoa(&c->addr), datetime_str(c->time)));
			GLOBAL_UNLOCK();
			return 0;
		}
	}
	GLOBAL_UNLOCK();
	errno = ENOENT;
	return -1;
}


/**
 * @brief Add a peer address into the list of peers we are trying to connect
 *
 * Add a peer address (and port) into the list of peers that
 * we are trying to connect. The timeout is an arbitrary
 * value of 60 seconds until we might again try to
 * connect that same address:port (unless it is blocked, too).
 *
 * @param addr pointer to a struct sockaddr_in of the peer
 *
 * @result 0 on succes, -1 and errno = EAGAIN if our list is full
 */
int peer_add_connecting(const struct sockaddr_in *addr)
{
	addr_time_t *c;
	time_t t0;
	FUN("peer_add_connecting");

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);
	if (g_peer->connect_cnt >= CONNECT_MAX) {
		LOGS(L_PEER,L_ERROR,("connect list is full (%d)\n",
			CONNECT_MAX));
		GLOBAL_UNLOCK();
		errno = EAGAIN;
		return -1;
	}
	c = &g_peer->connect[g_peer->connect_cnt];
	c->addr = *addr;
	c->time = t0 + 60;
	LOGS(L_PEER,L_MINOR,("won't try to connect entry #%d peer %s until %s\n",
		(int)g_peer->connect_cnt,
		sock_ntoa(&c->addr), datetime_str(c->time)));
	g_peer->connect_cnt += 1;
	GLOBAL_UNLOCK();
	return 0;
}

/**
 * @brief Check if a peer address is in the list of blocked peers
 *
 * Return zero if the peer specified in 'addr' is in the blocked list,
 * otherwise return -1 and set errno to ENOENT.
 * Blocked entries can have a specific port, or -1 (0xffff), which
 * means that all ports on this peer address are to be blocked.
 * If a blocking timed out, remove it from the list and move
 * down the list by one entry.
 *
 * @param addr pointer to a struct sockaddr_in of the peer
 *
 * @result zero if found, -1 and errno = ENOENT if not found
 */
int peer_blocked(const struct sockaddr_in *addr)
{
	addr_time_t *b;
	time_t t;
	size_t i, j;
	FUN("peer_blocked");

	if (0 != GLOBAL_LOCK())
		return -1;

	t = time(NULL);

	for (i = 0; i < g_peer->blocked_cnt; i++) {
		b = &g_peer->blocked[i];
		if (b->time < t) {
			LOGS(L_PEER,L_MINOR,("blocked entry # %d node %s is done (%s)\n",
				(int)i, sock_ntoa(&b->addr), datetime_str(b->time)));
			for (j = i; j < g_peer->blocked_cnt - 1; j++)
				g_peer->blocked[j] = g_peer->blocked[j+1];
			g_peer->blocked_cnt -= 1;
			i--;
			continue;
		}
		if (b->addr.sin_addr.s_addr != addr->sin_addr.s_addr) {
			continue;
		}
		if (-1 == (short)b->addr.sin_port) {
			LOGS(L_PEER,L_DEBUGX,("node %s:* is blocked until %s\n",
				inet_ntoa(b->addr.sin_addr), datetime_str(b->time)));
			GLOBAL_UNLOCK();
			return 0;
		}
		if (b->addr.sin_port == addr->sin_port) {
			LOGS(L_PEER,L_DEBUGX,("node %s is blocked until %s\n",
				sock_ntoa(&b->addr), datetime_str(b->time)));
			GLOBAL_UNLOCK();
			return 0;
		}
	}
	GLOBAL_UNLOCK();
	errno = ENOENT;
	return -1;
}

/**
 * @brief A peer address (and port) failed, see if it should be blocked
 * 
 * Add a addr:port into the list of failed connects. If it was already
 * there, add it into the blocked list for some time.
 *
 * @param addr pointer to a struct sockaddr_in of the peer
 *
 * @result zero on success, -1 on error (errno=EAGAIN if failure list is full)
 */
int peer_add_failure(const struct sockaddr_in *addr)
{
	addr_time_t *f, *b;
	time_t t0, d;
	size_t i, j;
	FUN("peer_add_failure");

	/* If this peer is already blocked, bail out */
	if (0 == peer_blocked(addr)) {
		return 0;
	}

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);

	for (i = 0; i < g_peer->failure_cnt; i++) {
		f = &g_peer->failure[i];
		if (f->time + 60 * 60 < t0) {
			LOGS(L_PEER,L_MINOR,("removing node %s at %s from failure list\n",
				sock_ntoa(&f->addr), datetime_str(f->time)));
			for (j = i; j < g_peer->failure_cnt - 1; j++)
				g_peer->failure[j] = g_peer->failure[j+1];
			g_peer->failure_cnt -= 1;
			i--;
			continue;
		}
		if (f->addr.sin_addr.s_addr != addr->sin_addr.s_addr)
			continue;
		if (-1 == (short)f->addr.sin_port || f->addr.sin_port == addr->sin_port) {
			d = t0 - f->time;
			LOGS(L_PEER,L_MINOR,("peer %s last failed at %s; block for %ds\n",
				sock_ntoa(&f->addr), datetime_str(f->time), (int)d));
			if (g_peer->blocked_cnt >= BLOCKED_MAX) {
				LOGS(L_PEER,L_MINOR,("blocked list is full (%d)\n",
					BLOCKED_MAX));
			} else {
				b = &g_peer->blocked[g_peer->blocked_cnt];
				b->addr = *addr;
				b->time = t0 + d;
				LOGS(L_PEER,L_MINOR,("blocked entry #%d peer %s until %s\n",
					(int)g_peer->blocked_cnt,
					sock_ntoa(&b->addr),
					datetime_str(b->time)));
				g_peer->blocked_cnt += 1;
			}
			GLOBAL_UNLOCK();
			return 0;
		}
	}

	if (g_peer->failure_cnt >= FAILURE_MAX) {
		LOGS(L_PEER,L_MINOR,("failure list is full (%d)\n",
			FAILURE_MAX));
		GLOBAL_UNLOCK();
		errno = EAGAIN;
		return -1;
	}
	f = &g_peer->failure[g_peer->failure_cnt];
	f->addr = *addr;
	f->time = t0;
	LOGS(L_PEER,L_MINOR,("failure entry #%d peer %s at %s\n",
		(int)g_peer->failure_cnt,
		sock_ntoa(&f->addr),
		datetime_str(f->time)));
	g_peer->failure_cnt += 1;
	GLOBAL_UNLOCK();
	return 0;
}

/**
 * @brief Set a MSG_FEC request key's status
 *
 * The list of FEC messages receives a new status
 * The index is built from the first two bytes of
 * the SHA1 digest.
 *
 * @param st Pointer to a fec_status_t
 *
 * @result zero on success (always!?)
 */
static int key_fec_status_set(fec_status_t *st)
{
	uint32_t idx;
	FUN("key_fec_status_set");
	/* index for fec status */
	idx = st->sha1.digest[0] | ((uint32_t)st->sha1.digest[1] << 8);
	idx >>= 32 - FECSTATUSSIZE_LOG2;

	g_peer->fec[idx] = *st;

	return 0;
}

/**
 * @brief Get a MSG_FEC request key's status
 *
 * Verify if the list of FEC messages for the
 * computed index is the right SHA1 digest,
 * and if so return its status.
 * Otherwise set the status to FEC_UNUSED
 * and retries to zero.
 *
 * @param st Pointer to a fec_status_t to be modified
 *
 * @result zero on success (always!?)
 */
static int key_fec_status_get(fec_status_t *st)
{
	uint32_t idx;
	FUN("key_fec_status_get");
	/* index for fec status */
	idx = st->sha1.digest[0] | ((uint32_t)st->sha1.digest[1] << 8);
	idx >>= 32 - FECSTATUSSIZE_LOG2;

	if (0 == memcmp(&g_peer->fec[idx].sha1, &st->sha1, SHA1SIZE)) {
		*st = g_peer->fec[idx];
		return 0;
	}
	st->status = FEC_UNUSED;
	st->retry = 0;
	return 0;
}

/**
 * @brief Check if a message is (probably) valid.
 *
 * Check if a message is probably valid. Probably, because
 * we can not look into encrypted keys without knowing their
 * encryption key - of course - so we just check the header
 * for sane looking data.
 *
 * A message can be valid in one of two ways:
 * 1. The given SHA1 hash matches the SHA1 of CHUNKSIZE bytes
 *    in msg->data
 * 2. The data has a valid key header and an allowable size
 *    of the (compressed) contents.
 */
int peer_msg_valid(peer_msg_t *msg)
{
	sha1_digest_t s;
	size_t size;
	FUN("peer_msg_valid");

	if (0 == memcmp(KEY_VERSION, msg->data, 4)) {
		size = ntohl(*(uint32_t*)&msg->data[4]);
		/* Allow smaller by 1 for now, because sometimes the last
		 * byte of an encrypted type 2 key is zero (who woulda thought?)
		 * TODO: Fix keys, so that they always end non 00 somehow
		 */
		if (msg->size < size - 1) {
			LOGS(L_PEER,L_ERROR,("invalid type 2 key (size:%#x, head:%#x)\n",
				msg->size, size));
			errno = EINVAL;
			return -1;
		}
		if (size >= 3*4 && size <= CHUNKSIZE && msg->size != size) {
			LOGS(L_PEER,L_MINOR,("different size:%#x, head:%#x\n",
				msg->size, size));
			msg->size = size;
			memset(&msg->data[size], 0, CHUNKSIZE - size);
		}
		if (size >= 3*4 && size <= CHUNKSIZE) {
			LOGS(L_PEER,L_DEBUG,("valid type 2 key (%#x) %s\n",
				(unsigned)size,
				sha1_hexshort(&msg->sha1)));
			return 0;
		}
	}

	sha1(msg->data, CHUNKSIZE, &s);

	if (0 == memcmp(&s, &msg->sha1, SHA1SIZE)) {
		/* valid chunk type 1. */
		LOGS(L_PEER,L_DEBUGX,("valid type 1\nsize:%#x key:%s",
			(unsigned)msg->size,
			sha1_hexshort(&msg->sha1)));
		return 0;
	}

	if (0 != memcmp(KEY_VERSION, msg->data, 4)) {
		LOGS(L_PEER,L_ERROR,("invalid type 2 key (not '%s')\n",
			KEY_VERSION));
		errno = EINVAL;
		return -1;
	}

	size = ntohl(*(uint32_t*)&msg->data[4]);

	if (size < 3*4) {
		LOGS(L_PEER,L_ERROR,("invalid type 2 key (too small)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	if (size >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid type 2 key (too big)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	return 0;
}

/**
 * @brief Advertize one or more keys and sleep for an adequate wait time
 *
 * Advertizes one or more keys with retries until the keys are
 * all inserted into at least one queue.
 *
 * @param s pointer to the list of SHA1 digests of the keys
 * @param count number of keys to advertize (1 to 12)
 * @param pflags pointer to flag bits per key (bit=0: adv, bit=1: done)
 * @param htl hops-to-live for the advertizement
 *
 * @result zero on success, -1 and errno set on error
 */
int peer_adv_wait(sha1_digest_t *s, size_t count, int *pflags, int htl)
{
	struct timeval t0, tv, to;
	int64_t timeout, limit;
	size_t i, n, keys = 0, adv, advsum = 0, required;
	int try, bit, sum, advflags, keyflags = *pflags;
	int routenotfound = 0, timedout = 0, rc = 0;
	FUN("peer_adv_wait");

	if (htl > MAX_HTL) {
		htl = MAX_HTL;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}
	if (0 == count || count > 16) {
		LOGS(L_PEER,L_ERROR,("count is invalid %d!\n", (int)count));
		errno = EINVAL;
		return -1;
	}

	required = count;

	/* start time */
	gettimeofday(&t0, NULL);

	/* timeout depends on available outbound bandwidth */
	limit = MAX(g_peer->out.available, g_conf->bwlimit_out / 32);
	sum = count - bitsum(keyflags);
	timeout = (int64_t)1000000 * htl * sum * MIN_MSGSIZE / limit;
	tv_add(&to, &t0, timeout);

	LOGS(L_PEER,L_DEBUG,("timeout for %d keys is %s\n",
		(int)count, usec_str(timeout)));

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	advflags = keyflags;
	advsum = 0;
	for (try = 0; try < (int)count; try++) {
		routenotfound = 0;
		timedout = 0;
		for (i = 0, keys = 0; i < count && keys < count; i++) {
			n = i;
			bit = 1 << n;
			if (0 != (keyflags & bit)) {
				keys++;
				continue;
			}
			if (htl <= 0) {
				keys++;
				advflags |= bit;
				continue;
			}
			if (0 == (advflags & bit)) {
				rc = peer_adv_key(&s[n], try, htl, &adv, 1);
				if (0 == rc) {
					/* advertized to at least one peer? */
					if (adv > 0) {
						keys++;
						advflags |= bit;
						advsum += adv;
					}
				} else if (EINVAL == errno) {
					return -1;
				} else if (EHOSTUNREACH == errno) {
					routenotfound = 1;
				} else if (ETIMEDOUT == errno) {
					timedout = 1;
				}
			}
		}
		if (keys >= required)
			break;
		gettimeofday(&tv, NULL);
		/* timeout ? */
		if (tv_cmp(&to, &tv) <= 0)
			break;
		timeout = tv_diff(&to, &tv) / 16;
		/* 50ms sleeps */
		if (timeout < 50000)
			timeout = 50000;
		osd_usleep(timeout);
		if (tv_cmp(&to, &tv) <= 0)
			break;
	}
	gettimeofday(&tv, NULL);
	timeout = tv_diff(&tv, &t0);
	LOGS(L_PEER,L_MINOR,("ADV %d tries took %s\n",
		++try, usec_str(timeout)));

	if (advsum > 0) {
		/*
		 * The timeout depends on available outbound bandwidth,
		 * minimum is (configured bandwidth)/8 to avoid very long
		 * delays here.
		 */
		limit = MAX(g_peer->out.available, g_conf->bwlimit_out / 32);
		timeout = (int64_t)1000000 * advsum * MIN_MSGSIZE / limit;
		timeout = MIN(timeout, (int64_t)60 * 1000000);
		tv_add(&to, &t0, timeout);

		/* continue the delay until timeout */
		gettimeofday(&tv, NULL);
		if (tv_cmp(&to, &tv) > 0) {
			timeout = MAX(tv_diff(&to, &tv), 50000);
			LOGS(L_PEER,L_MINOR,("sleep:%s keys:%d adv:%d HTL:%d avail:%llu\n",
				usec_str(timeout), (int)count, (int)advsum, (int)htl, (unsigned long long)limit));
			osd_usleep(timeout);
		}
	}

	if (keys < required) {
		LOGS(L_PEER,L_DEBUG,("adv %d of %d keys w/ HTL:%d, %d advsum\n",
			(int)keys, (int)count, htl, (int)advsum));
		timedout = 1;
	} else {
		/* reset intermediate error flags */
		routenotfound = 0;
		timedout = 0;
	}

	*pflags = keyflags | advflags;

	if (0 != routenotfound) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 != timedout) {
		errno = ETIMEDOUT;
		return -1;
	}

	return 0;
}

/**
 * @brief Request one or more keys and sleep for an adequate wait time
 *
 * Requests one or more keys with retries until the requests are
 * all inserted into at least one queue. The delay is guesstimated
 * based on the available inbound bandwidth, number of keys requested,
 * hops-to-live etc.
 * The *pflags indicate which keys are already available (bit = 1)
 * and which ones should be requested (bit = 0).
 * The *pflags are modified afterwards to represent the keys that
 * could be loaded from the data store at the end of the function.
 *
 * @param s pointer to the list of SHA1 digests of the keys
 * @param count number of keys to advertize (1 to 12)
 * @param pflags pointer to flag bits per key (bit=0: req, bit=1: done)
 * @param htl hops-to-live for the advertizement
 * @param buff pointer to a count times CHUNKSIZE buff to receive keys
 *
 * @result zero on success, -1 and errno set on error
 */
int peer_req_wait(sha1_digest_t *s, size_t count, int *pflags, int htl,
	uint8_t *buff)
{
	struct timeval t0, tv, to;
	int64_t timeout, limit;
	size_t i, n, delay, req, keys = 0, reqcnt = 0, reqsum = 0, required;
	int try, bit, sum, reqflags, keyflags = *pflags;
	int routenotfound = 0, timedout = 0, rc = 0;
	FUN("peer_req_wait");

	if (htl > MAX_HTL) {
		htl = MAX_HTL;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}
	if (0 == count || count > 16) {
		LOGS(L_PEER,L_ERROR,("count is invalid %d!\n", (int)count));
		errno = EINVAL;
		return -1;
	}

	required = count;
	if (12 == required)
		required = 8;

	/* start time */
	gettimeofday(&t0, NULL);

	/* Timeout depends on available inbound bandwidth */
	limit = MAX(g_peer->in.available, g_conf->bwlimit_in / 32);
	/* Number of chunks we might request */
	sum = count - bitsum(keyflags);
	/* Assume MAX_MSGSIZE messages (worst case) */
	timeout = (int64_t)1000000 * htl * sum * MAX_MSGSIZE / limit;
	/* But try no longer than 60 seconds */
	timeout = MIN(timeout, (int64_t)60 * 1000000);
	tv_add(&to, &t0, timeout);

	LOGS(L_PEER,L_DEBUG,("timeout for %d keys w/ HTL:%d is %s\n",
		(int)count, htl, usec_str(timeout)));

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	reqflags = keyflags;
	reqsum = 0;
	for (try = 0; try < (int)count; try++) {
		routenotfound = 0;
		timedout = 0;
		for (i = 0, keys = 0; i < count && keys < required; i++) {
			n = i;
			bit = 1 << n;
			if (0 != (keyflags & bit)) {
				keys++;
				continue;
			}
			rc = store_get(&s[n], &buff[n * CHUNKSIZE]);
			if (0 == rc) {
				LOGS(L_PEER,L_DEBUG,("found %s in the store\n",
					sha1_hexshort(&s[n])));
				keys++;
				keyflags |= bit;
				continue;
			}
			if (EINVAL == errno) {
				return -1;
			}
			if (htl <= 0) {
				timedout = 1;
				keys++;
				continue;
			}
			if (0 == (reqflags & bit)) {
				rc = peer_req_key(&s[n], try, htl, &req, 1);
				/* requested from at least one peer? */
				if (0 == rc) {
					if (req > 0) {
						reqflags |= bit;
						reqsum += req;
						reqcnt++;
					}
				} else if (EINVAL == errno) {
					return -1;
				} else if (EHOSTUNREACH == errno) {
					routenotfound = 1;
				} else if (ETIMEDOUT == errno) {
					timedout = 1;
				}
			}
		}
		if (keys >= required)
			break;
		if (htl <= 0) {
			timedout = 1;
			break;
		}
		gettimeofday(&tv, NULL);
		/* timeout ? */
		if (tv_cmp(&to, &tv) <= 0)
			break;
		timeout = tv_diff(&to, &tv) / 16;
		/* 50ms sleeps */
		if (timeout < 50000)
			timeout = 50000;
		osd_usleep(timeout);
		if (tv_cmp(&to, &tv) <= 0)
			break;
	}
	gettimeofday(&tv, NULL);
	timeout = tv_diff(&tv, &t0);
	LOGS(L_PEER,L_MINOR,("REQ %d tries took %s\n",
		++try, usec_str(timeout)));

	if (reqsum > 0) {
		/*
		 * The timeout depends on available inbound bandwidth,
		 * minimum is configured bandwidth/32 to avoid very long
		 * delays here.
		 */
		limit = MAX(g_peer->in.available, g_conf->bwlimit_in / 32);
		/* Assume MAX_MSGSIZE messages (worst case) */
		timeout = (int64_t)1000000 * reqcnt * MAX_MSGSIZE / limit;
		/* But sleep no longer than 60 seconds */
		timeout = MIN(timeout, (int64_t)60 * 1000000);
		tv_add(&to, &t0, timeout);

		LOGS(L_PEER,L_MINOR,("sleep:%s keys:%d req:%d HTL:%d avail:%llu\n",
			usec_str(timeout), (int)count, (int)reqsum, (int)htl, (unsigned long long)limit));
		/* check the store for keys 4 times, minimum sleep 50ms */
		delay = MAX(timeout / 4, 50000);
		do {
			/* all requested keys in the store now? */
			if (0 == store_check(s, reqflags, count, required))
				break;
			if (EINVAL == errno)
				return -1;
			osd_usleep(delay);
			gettimeofday(&tv, NULL);
		} while (tv_cmp(&to, &tv) > 0);
	}

	/* check what keys were requested but are not yet here */
	for (n = 0, keys = 0; n < count; n++) {
		bit = 1 << n;
		rc = store_get(&s[n], &buff[n * CHUNKSIZE]);
		if (0 == rc) {
			keys++;
			keyflags |= bit;
			continue;
		}
		/* zero memory for this key, since we don't have it */
		memset(&buff[n * CHUNKSIZE], 0, CHUNKSIZE);
	}

	if (keys < required) {
		LOGS(L_PEER,L_DEBUG,("got %d of %d keys w/ HTL:%d; %d requests\n",
			(int)keys, (int)count, htl, (int)reqsum));
		timedout = 1;
	} else {
		routenotfound = 0;
		timedout = 0;
		/* we can cancel all outstanding requests */
		for (n = 0; n < count; n++) {
			rc = peer_req_cancel(&s[n]);
			if (0 != rc && EINVAL == errno) {
				return -1;
			}
		}
	}

	*pflags = keyflags;

	if (0 != routenotfound) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 != timedout) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/**
 * @brief Inserts one or more keys and sleep for an adequate wait time
 *
 * Inserts one or more keys with retries until the keys are
 * all inserted into at least one queue. The delay is guesstimated
 * based on the available outbound bandwidth, number of keys inserted,
 * hops-to-live, sizeof of the keys etc.
 * The *pflags indicate which keys are already sent (bit = 1)
 * and which ones should be sent now (bit = 0).
 * The *pflags are modified afterwards to represent the keys that
 * could be queued and will (probably) go out to some node.
 *
 * @param s pointer to the list of SHA1 digests of the keys
 * @param count number of keys to insert (1 to 12)
 * @param pflags pointer to flag bits per key (bit=0: adv, bit=1: done)
 * @param htl hops-to-live for the advertizement
 * @param buff pointer to a count times CHUNKSIZE buff to receive keys
 *
 * @result zero on success, -1 and errno set on error
 */
int peer_ins_wait(sha1_digest_t *s, size_t count, int *pflags, int htl,
	uint8_t *buff)
{
	struct timeval t0, tv, to;
	int64_t timeout, limit;
	uint8_t *data;
	size_t sizes[16];
	size_t i, n, size, ins, keys = 0, inssum = 0, inssize = 0, required;
	int try, bit, insflags, keyflags = *pflags;
	int routenotfound = 0, timedout = 0, rc = 0;
	FUN("peer_ins_wait");

	if (htl > MAX_HTL) {
		htl = MAX_HTL;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}
	if (0 == count || count > 16) {
		LOGS(L_PEER,L_ERROR,("count is invalid %d!\n", (int)count));
		errno = EINVAL;
		return -1;
	}

	for (n = 0; n < count; n++) {
		size = CHUNKSIZE;
		data = &buff[n * CHUNKSIZE];
		while (size > 1 && 0 == data[size-1])
			size--;
		sizes[n] = size;
	}

	required = count;
	if (12 == required)
		required = 8;

	/* start time */
	gettimeofday(&t0, NULL);

	/* Timeout depends on available outbound bandwidth */
	limit = MAX(g_peer->out.available, g_conf->bwlimit_out / 32);
	/* The number of bytes to insert preset the timeout */
	timeout = (int64_t)1000000 * inssize / limit;
	/* But try no longer than 120 seconds */
	timeout = MIN(timeout, (int64_t)120 * 1000000);
	tv_add(&to, &t0, timeout);

	if (htl > 0) {
		LOGS(L_PEER,L_DEBUG,("timeout for %d keys %#x is %s\n",
			(int)count, (unsigned)inssize, usec_str(timeout)));
	}

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	insflags = 0;
	inssum = 0;
	inssize = 0;
	for (try = 0; try < (int)count; try++) {
		routenotfound = 0;
		timedout = 0;
		for (i = 0, keys = 0; i < count && keys < count; i++) {
			n = i;
			bit = 1 << n;
			if (0 != (keyflags & bit)) {
				insflags |= bit;
				keys++;
				continue;
			}
			if (htl <= 0) {
				insflags |= bit;
				keys++;
				continue;
			}
			if (0 == (insflags & bit)) {
				data = &buff[n * CHUNKSIZE];
				size = sizes[n];
				rc = peer_ins_key(&s[n], try, htl, &ins, size, data, 1);
				/* inserted to at least one peer? */
				if (0 == rc) {
					if (ins > 0) {
						keys++;
						insflags |= bit;
						inssum += ins;
						inssize += ins * size;
					}
				} else if (EINVAL == errno) {
					return -1;
				} else if (EHOSTUNREACH == errno) {
					routenotfound = 1;
				} else if (ETIMEDOUT == errno) {
					timedout = 1;
				}
			}
		}
		if (keys >= required)
			break;
		gettimeofday(&tv, NULL);
		/* timeout ? */
		if (tv_cmp(&to, &tv) <= 0)
			break;
		timeout = tv_diff(&to, &tv) / 16;
		/* 50ms sleeps */
		if (timeout < 50000)
			timeout = 50000;
		osd_usleep(timeout);
		if (tv_cmp(&to, &tv) <= 0)
			break;
	}
	gettimeofday(&tv, NULL);
	timeout = tv_diff(&tv, &t0);
	LOGS(L_PEER,L_MINOR,("INS %d tries took %s\n",
		++try, usec_str(timeout)));

	if (inssum > 0) {
		/*
		 * The timeout depends on available outbound bandwidth,
		 * minimum is configured bandwidth/32 to avoid very long
		 * delays here.
		 */
		limit = MAX(g_peer->out.available, g_conf->bwlimit_out / 32);
		/* The number of bytes to insert presets the timeout */
		timeout = (uint64_t)1000000 * inssize / limit;
		/* But sleep no longer than 120 seconds */
		timeout = MIN(timeout, (int64_t)120 * 1000000);
		tv_add(&to, &t0, timeout);
		gettimeofday(&tv, NULL);
		if (tv_cmp(&to, &tv) > 0) {
			timeout = MAX(tv_diff(&to, &tv), 50000);
			LOGS(L_PEER,L_MINOR,("sleep:%s keys:%d ins:%d HTL:%d avail:%llu\n",
				usec_str(timeout), (int)count, (int)inssum, (int)htl, (unsigned long long)limit));
			osd_usleep(timeout);
		}
	}

	if (keys < required) {
		LOGS(L_PEER,L_DEBUG,("put %d of %d keys w/ HTL:%d; %d inserts\n",
			(int)keys, (int)count, htl, (int)inssum));
		timedout = 1;
	} else {
		/* reset intermediate error flags */
		routenotfound = 0;
		timedout = 0;
	}

	*pflags = keyflags | insflags;

	if (0 != routenotfound) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 != timedout) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/*
 *	peer_adv_key()
 *	Advertizes a chunk of data with the sha1 digest in 's' to a
 *	maximum of 'max' peer's queues. If a peer is interested in the
 *	data, it will then request it. The number of queued advertizements
 *	out is returned in 'adv'.
 */
int peer_adv_key(sha1_digest_t *s, int try, int htl, size_t *adv, int local)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j, k, l, fpr, advcnt;
	size_t seq[PEERS_OUT_MAX], best_queue, qs;
	uint64_t best_store, ss;
	uint32_t bit, seqflags, thr, spr, best_thr;
	int route;
	FUN("peer_adv_key");

	if (htl <= 0) {
		LOGS(L_PEER,L_DEBUG,("HTL is %d\n", htl));
		errno = EINVAL;
		return -1;
	}

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	} else if (htl < 2) {
		htl = 2;
		LOGS(L_PEER,L_DEBUG,("HTL upped to %d!\n", htl));
	}

	thr = threshold_for_htl(try, htl);
	spr = spread_for_htl(htl);

	/* which flavor of key is this? */
	fpr = keyroute(s);

	if (0 != GLOBAL_LOCK())
		return -1;

	if ((j = g_peer->peercnt_out) < 1) {
		GLOBAL_UNLOCK();
		errno = EAGAIN;
		return -1;
	}

	/* find ascending sequence of low queue sizes, followed by empty stores */
	for (i = 0, seqflags = 0; i < j; i++) {
		best_thr = 0;
		best_queue = MSG_QUEUESIZE;
		best_store = 0;
		seq[i] = i;
		for (k = 0; k < j; k++) {
			bit = 1 << k;
			if (0 != (seqflags & bit))
				continue;
			if (NULL == (p = g_peer->peers_out[i])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			q = &p->queue;
			/* free space in store */
			if (p->storesize > 0 && p->currsize < p->storesize)
				ss = p->storesize - p->currsize;
			else
				ss = 0;
			/* current queue size */
			qs = QUEUE_SIZE(q);

			if (p->fingerprint[fpr] > best_thr) {
				/* best fingerprint threshold so far */
				best_thr = p->fingerprint[fpr];
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs < best_queue) {
				/* empty queue slots for this peer is best so far */
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs == best_queue && ss > best_store) {
				/* if queue size is the same, prefer peers with empty store */
				best_store = ss;
				seq[i] = k;
			}
		}
		bit = 1 << seq[i];
		seqflags |= bit;
	}

	advcnt = 0;
	route = 0;
	best_thr = 0;
	for (i = 0; i < j; i++) {
		l = seq[i];
		if (NULL == (p = g_peer->peers_out[l])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->fingerprint[fpr] > best_thr) {
			/* remember best threshold */
			best_thr = p->fingerprint[fpr];
		}
		if (p->fingerprint[fpr] < thr) {
			/* fingerprint below threshold */
			continue;
		}
		/* the route would be there.. keep this in mind */
		route = 1;

		q = &p->queue;
		qs = QUEUE_SIZE(q);
		if (0 == local && qs >= ADV_QUEUELIMIT) {
			/* outbound queue is full */
			continue;
		}
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		/* search for a queue entry for this message */
		for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
			e = &q->entry[k];
			if (e->cmd == MSG_ADV &&
				0 == memcmp(&e->sha1, s, SHA1SIZE))
				break;
		}
		if (k != q->head) {
			/* found a queue entry */
			advcnt += 1;
		} else {
			e = &q->entry[q->head];
			e->cmd = MSG_ADV;
			e->htl = htl;
			e->sha1 = *s;
			e->size = 0;
			QUEUE_NEXT(q);
			advcnt += 1;
			LOGS(L_PEER,L_DEBUG,("ADV %s to %s on queue #%d after %d\n",
				sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
		}
		if (advcnt >= spr)
			break;
	}

	/* Try again with best_thr if no route was found */
	if (best_thr > 0 && (0 == route || 0 == advcnt)) {
		for (i = 0; i < j; i++) {
			l = seq[i];
			if (NULL == (p = g_peer->peers_out[l])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			if (p->fingerprint[fpr] < best_thr) {
				/* fingerprint below threshold */
				continue;
			}
			/* the route would be there.. keep this in mind */
			route = 1;

			q = &p->queue;
			qs = QUEUE_SIZE(q);
			if (0 == local && qs >= ADV_QUEUELIMIT) {
				/* outbound queue is full */
				continue;
			}
			if (0 != QUEUE_FULL(q)) {
				/* outbound queue is full */
				continue;
			}

			/* search for a queue entry for this message */
			for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
				e = &q->entry[k];
				if (e->cmd == MSG_ADV &&
					0 == memcmp(&e->sha1, s, SHA1SIZE))
					break;
			}
			if (k != q->head) {
				/* found a queue entry */
				advcnt += 1;
			} else {
				e = &q->entry[q->head];
				e->cmd = MSG_ADV;
				e->htl = htl;
				e->sha1 = *s;
				e->size = 0;
				QUEUE_NEXT(q);
				advcnt += 1;
				LOGS(L_PEER,L_DEBUG,("ADV %s to %s on queue #%d after %d\n",
					sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
			}
			if (advcnt >= spr)
				break;
		}
	}
	GLOBAL_UNLOCK();

	if (NULL != adv) {
		*adv = advcnt;
	}
	if (0 == route) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 == advcnt) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/*
 *	peer_req_key()
 *	Request a chunk of data with the SHA1 hash 's' from a
 *	maxmimum of 'max' peer's queues. The number of queued requests
 *	is returned in 'req'
 */
int peer_req_key(sha1_digest_t *s, int try, int htl, size_t *req, int local)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j, k, l, fpr, reqcnt;
	size_t seq[PEERS_OUT_MAX], best_queue, qs;
	uint64_t best_store, ss;
	uint32_t bit, seqflags, thr, spr, best_thr;
	int route;
	FUN("peer_req_key");

	if (htl <= 0) {
		LOGS(L_PEER,L_DEBUG,("HTL is %d\n", htl));
		errno = EINVAL;
		return -1;
	}

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	} else if (htl < 2) {
		htl = 2;
		LOGS(L_PEER,L_DEBUG,("HTL upped to %d!\n", htl));
	}

	thr = threshold_for_htl(try, htl);
	spr = spread_for_htl(htl);

	/* which flavor of key is this? */
	fpr = keyroute(s);

	if (0 != GLOBAL_LOCK())
		return -1;

	if ((j = g_peer->peercnt_out) < 1) {
		GLOBAL_UNLOCK();
		errno = EAGAIN;
		return -1;
	}

	/* find ascending sequence of low queue sizes, followed by full stores */
	for (i = 0, seqflags = 0; i < j; i++) {
		best_thr = 0;
		best_queue = MSG_QUEUESIZE;
		best_store = 0;
		seq[i] = i;
		for (k = 0; k < j; k++) {
			bit = 1 << k;
			if (0 != (seqflags & bit))
				continue;
			if (NULL == (p = g_peer->peers_out[i])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			q = &p->queue;
			/* used space in store */
			if (p->storesize > 0 && p->currsize < p->storesize)
				ss = p->currsize;
			else
				ss = 0;
			qs = QUEUE_SIZE(q);
			if (p->fingerprint[fpr] > best_thr) {
				/* best fingerprint threshold so far */
				best_thr = p->fingerprint[fpr];
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs < best_queue) {
				/* empty queue slots for this peer is best so far */
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs == best_queue && ss > best_store) {
				/* if queue size is the same, prefer peers with full store */
				best_store = ss;
				seq[i] = k;
			}
		}
		bit = 1 << seq[i];
		seqflags |= bit;
	}

	reqcnt = 0;
	route = 0;
	best_thr = 0;
	for (i = 0; i < j; i++) {
		l = seq[i];
		if (NULL == (p = g_peer->peers_out[l])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->fingerprint[fpr] > best_thr) {
			/* remember best threshold */
			best_thr = p->fingerprint[fpr];
		}
		if (p->fingerprint[fpr] < thr) {
			/* fingeprint below threshold */
			continue;
		}

		/* the route would be there.. keep this in mind */
		route = 1;

		q = &p->queue;
		qs = QUEUE_SIZE(q);
		if (0 == local && qs >= REQ_QUEUELIMIT) {
			/* outbound queue is full */
			continue;
		}
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		/* search for a queue entry for this message */
		for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
			e = &q->entry[k];
			if (e->cmd == MSG_REQ &&
				0 == memcmp(&e->sha1, s, SHA1SIZE))
				break;
		}
		if (k != q->head) {
			/* found a queue entry */
			reqcnt += 1;
		} else {
			e = &q->entry[q->head];
			e->cmd = MSG_REQ;
			e->htl = htl;
			e->sha1 = *s;
			e->size = 0;
			QUEUE_NEXT(q);
			reqcnt += 1;
			LOGS(L_PEER,L_DEBUG,("REQ %s from %s on queue #%d after %d\n",
				sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
		}
		if (reqcnt >= spr)
			break;
	}

	/* Try again with best_thr if no route was found */
	if (best_thr > 0 && (0 == route || 0 == reqcnt)) {
		for (i = 0; i < j; i++) {
			l = seq[i];
			if (NULL == (p = g_peer->peers_out[l])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			if (p->fingerprint[fpr] < best_thr) {
				/* fingeprint below threshold */
				continue;
			}

			/* the route would be there.. keep this in mind */
			route = 1;

			q = &p->queue;
			qs = QUEUE_SIZE(q);
			if (0 == local && qs >= REQ_QUEUELIMIT) {
				/* outbound queue is full */
				continue;
			}
			if (0 != QUEUE_FULL(q)) {
				/* outbound queue is full */
				continue;
			}

			/* search for a queue entry for this message */
			for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
				e = &q->entry[k];
				if (e->cmd == MSG_REQ &&
					0 == memcmp(&e->sha1, s, SHA1SIZE))
					break;
			}
			if (k != q->head) {
				/* found a queue entry */
				reqcnt += 1;
			} else {
				e = &q->entry[q->head];
				e->cmd = MSG_REQ;
				e->htl = htl;
				e->sha1 = *s;
				e->size = 0;
				QUEUE_NEXT(q);
				reqcnt += 1;
				LOGS(L_PEER,L_DEBUG,("REQ %s from %s on queue #%d after %d\n",
					sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
			}
			if (reqcnt >= spr)
				break;
		}
	}
	GLOBAL_UNLOCK();

	if (NULL != req) {
		*req = reqcnt;
	}
	if (0 == route) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 == reqcnt) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/*
 *	peer_ins_key()
 *	Insert a chunk of data with the SHA1 hash 's' into a maximum
 *	of 'spr' peer's queues (spr is derived from htl).
 *	The number of queued inserts is returned in 'ins'.
 */
int peer_ins_key(sha1_digest_t *s, int try, int htl, size_t *ins,
	size_t size, uint8_t *buff, int local)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j, k, l, fpr, inscnt;
	size_t seq[PEERS_OUT_MAX], best_queue, qs;
	uint64_t best_store, ss;
	uint32_t bit, seqflags, thr, spr, best_thr;
	int route;
	FUN("peer_ins_key");

	if (htl <= 0) {
		LOGS(L_PEER,L_DEBUG,("HTL is %d\n", htl));
		errno = EINVAL;
		return -1;
	}

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	thr = threshold_for_htl(try, htl);
	spr = spread_for_htl(htl);

	/* which flavor of key is this? */
	fpr = keyroute(s);

	if (0 != GLOBAL_LOCK())
		return -1;

	if ((j = g_peer->peercnt_out) < 1) {
		GLOBAL_UNLOCK();
		errno = EAGAIN;
		return -1;
	}

	/* find ascending sequence of low queue sizes, followed by empty stores */
	for (i = 0, seqflags = 0; i < j; i++) {
		best_thr = 0;
		best_queue = MSG_QUEUESIZE;
		best_store = 0;
		seq[i] = i;
		for (k = 0; k < j; k++) {
			bit = 1 << k;
			if (0 != (seqflags & bit))
				continue;
			if (NULL == (p = g_peer->peers_out[i])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			q = &p->queue;
			if (p->storesize > 0 && p->currsize < p->storesize)
				ss = p->storesize - p->currsize;
			else
				ss = 0;
			qs = QUEUE_SIZE(q);
			if (p->fingerprint[fpr] > best_thr) {
				/* best fingerprint threshold so far */
				best_thr = p->fingerprint[fpr];
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs < best_queue) {
				/* empty queue slots for this peer is best so far */
				best_queue = qs;
				best_store = ss;
				seq[i] = k;
			} else if (qs == best_queue && ss > best_store) {
				/* if queue size is the same, prefer peers with empty store */
				best_store = ss;
				seq[i] = k;
			}
		}
		bit = 1 << seq[i];
		seqflags |= bit;
	}

	inscnt = 0;
	route = 0;
	best_thr = 0;
	for (i = 0; i < j; i++) {
		l = seq[i];
		if (NULL == (p = g_peer->peers_out[l])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->fingerprint[fpr] > best_thr) {
			/* remember best threshold */
			best_thr = p->fingerprint[fpr];
		}
		if (p->fingerprint[fpr] < thr) {
			/* fingerprint below threshold */
			continue;
		}

		/* the route would be there.. keep this in mind */
		route = 1;

		q = &p->queue;
		qs = QUEUE_SIZE(q);
		if (0 == local && qs >= INS_QUEUELIMIT) {
			/* outbound queue is full */
			continue;
		}
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		/* search for a queue entry for this message */
		for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
			e = &q->entry[k];
			if (e->cmd == MSG_INS &&
				0 == memcmp(&e->sha1, s, SHA1SIZE))
				break;
		}
		if (k != q->head) {
			/* found a queue entry */
			inscnt += 1;
		} else {
			e = &q->entry[q->head];
			e->cmd = MSG_INS;
			e->htl = htl;
			e->sha1 = *s;
			e->size = size;
			if (e->size <= MSG_BUFFSIZE) {
				/* message data fits in buffer */
				memcpy(e->data, buff, size);
			} else {
				/* fetch data from file just in time */
				memset(e->data, 0, MSG_BUFFSIZE);
			}
			QUEUE_NEXT(q);
			inscnt += 1;
			LOGS(L_PEER,L_DEBUG,("INS %s to %s on queue #%d after %d\n",
				sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
		}
		if (inscnt >= spr)
			break;
	}

	/* Try again with best_thr if no route was found */
	if (best_thr > 0 && (0 == route || 0 == inscnt)) {
		for (i = 0; i < j; i++) {
			l = seq[i];
			if (NULL == (p = g_peer->peers_out[l])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}
			if (p->fingerprint[fpr] < best_thr) {
				/* fingerprint below threshold */
				continue;
			}

			/* the route would be there.. keep this in mind */
			route = 1;

			q = &p->queue;
			qs = QUEUE_SIZE(q);
			if (0 == local && qs >= INS_QUEUELIMIT) {
				/* outbound queue is full */
				continue;
			}
			if (0 != QUEUE_FULL(q)) {
				/* outbound queue is full */
				continue;
			}

			/* search for a queue entry for this message */
			for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
				e = &q->entry[k];
				if (e->cmd == MSG_INS &&
					0 == memcmp(&e->sha1, s, SHA1SIZE))
					break;
			}
			if (k != q->head) {
				/* found a queue entry */
				inscnt += 1;
			} else {
				e = &q->entry[q->head];
				e->cmd = MSG_INS;
				e->htl = htl;
				e->sha1 = *s;
				e->size = size;
				if (e->size <= MSG_BUFFSIZE) {
					/* message data fits in buffer */
					memcpy(e->data, buff, size);
				} else {
					/* fetch data from file just in time */
					memset(e->data, 0, MSG_BUFFSIZE);
				}
				QUEUE_NEXT(q);
				inscnt += 1;
				LOGS(L_PEER,L_DEBUG,("INS %s to %s on queue #%d after %d\n",
					sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
			}
			if (inscnt >= spr)
				break;
		}
	}
	GLOBAL_UNLOCK();

	if (NULL != ins) {
		*ins = inscnt;
	}
	if (0 == route) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 == inscnt) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/*
 *	peer_req_fec()
 *	Create a FEC reconstruction request message (MSG_FEC) for a list
 *	of 13 hashes: 1 hash of the resulting fragment, 12 hashes of chunks.
 */
int peer_req_fec(sha1_digest_t *s, int try, int htl, size_t *fec, int local)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j, k, n, spr, feccnt;
	size_t hashsize;
	size_t offs;
	size_t size;
	size_t qs;
	uint8_t *buff = NULL;
	uint8_t routes[12];	/* routes of the 12 chunk SHA1 hashes */
	uint32_t thr, avg_thr, best_thr;
	int route;
	FUN("peer_req_fec");

	if (htl <= 0) {
		LOGS(L_PEER,L_DEBUG,("HTL is %d\n", htl));
		errno = EINVAL;
		return -1;
	}

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	/* get 1/4th of the threshold for this HTL value */
	thr = threshold_for_htl(try, htl) / 4;

	/* size of the 'expanded' buffer, i.e. the net data */
	hashsize = 12 * SHA1SIZE;

	/* size of the 'squeezed' buffer:
	 * 3 header values (version, size, hashsize) with 4 bytes each
	 * 1 LZW unsqueezed 0x00 flag
	 * 12 SHA1 hashes (size)
	 * 1 trailing byte
	 */
	size = 4 + 4 + 4 + 1 + hashsize + 1;

	buff = xcalloc(size, sizeof(uint8_t));
	offs = 0;

	/* fill buff with the KEY_VERSION and length values */
	buff[offs++] = KEY_VERSION[0];
	buff[offs++] = KEY_VERSION[1];
	buff[offs++] = KEY_VERSION[2];
	buff[offs++] = KEY_VERSION[3];

	/* compare this buffer layout with src/file.c file_squeeze() */
	*(uint32_t *)&buff[offs] = htonl(size);
	offs += 4;

	*(uint32_t *)&buff[offs] = htonl(hashsize);
	offs += 4;

	buff[offs++] = 0x00;	/* uncompressed */

	/* fill in the 12 hashes from s[1] to s[12] */
	for (i = 0; i < 12; i++) {
		memcpy(&buff[offs], &s[1+i], SHA1SIZE);
		routes[i] = keyroute(&s[1+i]);
		offs += SHA1SIZE;
	}
	/* terminate with a 0x55 byte in case the last SHA1 last byte(s) are 0x00 */
	buff[offs++] = 0x55;

	if (0 != GLOBAL_LOCK()) {
		xfree(buff);
		errno = EINVAL;
		return -1;
	}

	if ((j = g_peer->peercnt_out) < 1) {
		GLOBAL_UNLOCK();
		xfree(buff);
		errno = EAGAIN;
		return -1;
	}

	feccnt = 0;
	route = 0;
	best_thr = 0;
	spr = j / 8 + 1;
	/* spread to no more than two peers if this is no local request */
	if (0 == local && spr > 2)
		spr = 2;
	for (i = 0; i < j; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}

		/* average of fingerprint values for the 12 hashes */
		avg_thr = 0;
		for (n = 0; n < 12; n++)
			avg_thr += p->fingerprint[routes[n]];
		avg_thr /= 12;

		/* remember best threshold */
		if (avg_thr > best_thr) {
			best_thr = avg_thr;
		}

		/* skip this node if the average is below HTL's threshold */
		if (avg_thr < thr) {
			continue;
		}

		/* the route would be there.. keep this in mind */
		route = 1;

		q = &p->queue;
		qs = QUEUE_SIZE(q);
		if (0 == local && qs >= FEC_QUEUELIMIT) {
			/* outbound queue is full */
			continue;
		}
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		/* search for a queue entry for this message */
		for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
			e = &q->entry[k];
			if (e->cmd == MSG_FEC &&
				0 == memcmp(&e->sha1, s, SHA1SIZE))
				break;
		}
		if (k != q->head) {
			/* found a queue entry */
			feccnt += 1;
		} else {
			e = &q->entry[q->head];
			e->cmd = MSG_FEC;
			e->htl = htl;
			e->sha1 = s[0];
			e->size = size;
			if (e->size <= MSG_BUFFSIZE) {
				/* message data fits in buffer */
				memcpy(e->data, buff, size);
			} else {
				die(1, "Fatal: size > MSG_BUFFSIZE\n");
			}
			QUEUE_NEXT(q);
			feccnt += 1;
			LOGS(L_PEER,L_DEBUG,("FEC %s to %s on queue #%d after %d\n",
				sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
		}
		if (feccnt >= spr)
			break;
	}

	/* Final try with best_thr if no route was found */
	if (best_thr > 0 && (0 == route || 0 == feccnt)) {
		for (i = 0; i < j; i++) {
			if (NULL == (p = g_peer->peers_out[i])) {
				/* unused slot */
				continue;
			}
			if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
				/* not connected */
				continue;
			}

			/* average of fingerprint values for the 12 hashes */
			avg_thr = 0;
			for (n = 0; n < 12; n++)
				avg_thr += p->fingerprint[routes[n]];
			avg_thr /= 12;

			/* skip this node if the average is below best threshold */
			if (avg_thr < best_thr) {
				continue;
			}

			/* the route would be there.. keep this in mind */
			route = 1;

			q = &p->queue;
			qs = QUEUE_SIZE(q);
			if (0 == local && qs >= FEC_QUEUELIMIT) {
				/* outbound queue is full */
				continue;
			}
			if (0 != QUEUE_FULL(q)) {
				/* outbound queue is full */
				continue;
			}

			/* search for a queue entry for this message */
			for (k = q->tail; k != q->head; k = (k + 1) % q->size) {
				e = &q->entry[k];
				if (e->cmd == MSG_FEC &&
					0 == memcmp(&e->sha1, &sha1, SHA1SIZE))
					break;
			}
			if (k != q->head) {
				/* found a queue entry */
				feccnt += 1;
			} else {
				e = &q->entry[q->head];
				e->cmd = MSG_FEC;
				e->htl = htl;
				e->sha1 = s[0];
				e->size = size;
				if (e->size <= MSG_BUFFSIZE) {
					/* message data fits in buffer */
					memcpy(e->data, buff, size);
				} else {
					die(1, "Fatal: size > MSG_BUFFSIZE\n");
				}
				QUEUE_NEXT(q);
				feccnt += 1;
				LOGS(L_PEER,L_DEBUG,("FEC %s to %s on queue #%d after %d\n",
					sha1_hexshort(s), p->conn.peeraddr, (int)i, (int)qs));
			}
			if (feccnt >= spr)
				break;
		}
	}
	GLOBAL_UNLOCK();
	xfree(buff);

	if (NULL != fec) {
		*fec = feccnt;
	}
	if (0 == route) {
		errno = EHOSTUNREACH;
		return -1;
	}
	if (0 == feccnt) {
		errno = ETIMEDOUT;
		return -1;
	}
	return 0;
}

/*
 *	peer_adv_cancel()
 *	Cancel advertizements for a certain SHA1 to a peer before they go out
 *	of the queue. This is called whenever a key was inserted from a peer.
 */
static int peer_adv_cancel(sha1_digest_t *s, struct sockaddr_in *peeraddr)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j;
	int can = 0, rc = 0;
	FUN("peer_adv_cancel");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != peeraddr->sin_addr.s_addr ||
			p->conn.address.sin_port != peeraddr->sin_port) {
			/* not the node we want */
			continue;
		}
		/* check if this key is already queued */
		q = &p->queue;
		for (j = 0; j < q->size; j++) {
			e = &q->entry[j];
			if (MSG_ADV == e->cmd &&
				0 == memcmp(&e->sha1, s, SHA1SIZE)) {
				e->cmd = MSG_CAN;
				e->size = 0;
				can++;
			}
		}
	}
	GLOBAL_UNLOCK();

	if (0 == rc && can > 0) {
		LOGS(L_PEER,L_DEBUG,("cancelled %d pending advertizements for key %s\n",
			can, sha1_hexshort(s)));
	}
	return rc;
}

/*
 *	peer_req_cancel()
 *	Cancel all requests for a certain key before they go out of the queue.
 *	This is called whenever a key was successfully received and inserted.
 */
static int peer_req_cancel(sha1_digest_t *s)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j;
	int can = 0, rc = 0;
	FUN("peer_req_cancel");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED !=p->conn.status) {
			/* not connected */
			continue;
		}

		/* check if this key is already queued */
		q = &p->queue;
		for (j = 0; j < q->size; j++) {
			e = &q->entry[j];
			if (MSG_REQ == e->cmd &&
				0 == memcmp(&e->sha1, s, SHA1SIZE)) {
				e->cmd = MSG_CAN;
				e->size = 0;
				can++;
			}
		}
	}
	GLOBAL_UNLOCK();

	if (0 == rc && can > 0) {
		LOGS(L_PEER,L_DEBUG,("cancelled %d pending requests for key %s\n",
			can, sha1_hexshort(s)));
	}

	return rc;
}

/*
 *	peer_ins_cancel()
 *	Cancel inserts for a certain key to a peer before they go out of the
 *	queue. This is called whenever a key was advertized from a peer.
 */
static int peer_ins_cancel(sha1_digest_t *s, struct sockaddr_in *peeraddr)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	size_t i, j;
	int can = 0, rc = 0;
	FUN("peer_ins_cancel");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED !=p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != peeraddr->sin_addr.s_addr ||
			p->conn.address.sin_port != peeraddr->sin_port) {
			/* not the node we want */
			continue;
		}

		/* check if this key is already queued */
		q = &p->queue;
		for (j = 0; j < q->size; j++) {
			e = &q->entry[j];
			if (MSG_INS == e->cmd &&
				0 == memcmp(&e->sha1, s, SHA1SIZE)) {
				e->cmd = MSG_CAN;
				e->size = 0;
				can++;
			}
		}
	}
	GLOBAL_UNLOCK();

	if (0 == rc && can > 0) {
		LOGS(L_PEER,L_DEBUG,("cancelled %d pending inserts for key %s\n",
			can, sha1_hexshort(s)));
	}
	return rc;
}

/*
 *	peer_info()
 *	Return some info about then n'th peer status.
 */
int peer_info(size_t i, peer_info_t *info)
{
	peer_node_t *p;
	FUN("peer_info");

	if (i >= PEERS_MAX) {
		LOGS(L_PEER,L_DEBUG,("index (%#x) out of range (0..%#x)\n",
			i, (unsigned)(PEERS_MAX - 1)));
		errno = ENOENT;
		return -1;
	}

	memset(info, 0, sizeof(*info));
	BWLIMIT_LOCK();
	info->in_total = g_peer->in.total;
	info->out_total = g_peer->out.total;
	info->in_daily = g_peer->in.daily;
	info->out_daily = g_peer->out.daily;
	BWLIMIT_UNLOCK();

	if (i < PEERS_IN_MAX) {
		p = g_peer->peers_in[i];
	} else {
		p = g_peer->peers_out[i - PEERS_IN_MAX];
	}
	if (NULL != p) {
		if (0 == GLOBAL_LOCK()) {
			info->active = p->pid != 0;
			info->status = p->conn.status;
			strncpy(info->type, p->type, sizeof(info->type));
			info->type[sizeof(info->type)-1] = '\0';
			info->major = p->major;
			info->minor = p->minor;
			info->build = p->build;
			info->crypto_module_out = p->crypto_module_out;
			info->crypto_module_in = p->crypto_module_in;
			info->start = p->start;
			info->last_io = p->conn.last_io;
			info->address = p->conn.address;
			strncpy(info->peername, p->conn.peername,
				sizeof(info->peername));
			info->peername[sizeof(info->peername) - 1] = '\0';
			info->in_peer = p->conn.in_total;
			info->out_peer = p->conn.out_total;
			memcpy(info->fingerprint, p->fingerprint,
				sizeof(info->fingerprint));
			info->storesize = p->storesize;
			info->currsize = p->currsize;
			info->stats = p->stats;
			info->keycount = p->keycount;
			if (p->queue.size > 0)
				info->queue = p->queue;
			else
				info->queue.size = 0;
			GLOBAL_UNLOCK();
		}
	}

	return 0;
}

/*
 *	peer_load()
 *	Return some info about the current load in percent (0 to 100)
 */
int peer_load(peer_load_t *pl)
{
	peer_node_t *p;
	msg_queue_t *q;
	size_t queues_sum;
	size_t last_io_sum;
	size_t peer_in_count;
	size_t peer_out_count;
	size_t peer_count;
	time_t t0;
	double load;
	size_t i;
	FUN("peer_load");

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);
	queues_sum = 0;
	last_io_sum = 0;
	peer_in_count = 0;
	peer_out_count = 0;
	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (t0 >= p->conn.last_io) {
			time_t diff = t0 - p->conn.last_io;
			if (diff > 30) {
				diff = 30;
			}
			last_io_sum += 30 - diff;
		}
		peer_in_count++;
	}
	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (t0 >= p->conn.last_io) {
			time_t diff = t0 - p->conn.last_io;
			if (diff > 30) {
				diff = 30;
			}
			last_io_sum += 30 - diff;
		}
		/* outgoing connection */
		peer_out_count++;
		q = &p->queue;
		queues_sum += QUEUE_SIZE(q);
	}
	GLOBAL_UNLOCK();

	peer_count = peer_in_count + peer_out_count;
	load = 0.05;
	if (peer_count > 0) {
		load += 0.30 * last_io_sum / 30 / peer_count;
		load += 0.15 * peer_count / PEERS_MAX;
		if (peer_out_count > 0)
			load = load + 0.50 * queues_sum / MSG_QUEUESIZE / peer_out_count;
	}
	LOGS(L_PEER,L_MINOR,("load %d%% (peers:%d/%d, last_io:%d/%d, queues:%d/%d)\n",
		(int)(load * 100),
		(int)peer_count, PEERS_MAX,
		(int)last_io_sum, (int)(peer_count * 60),
		(int)queues_sum, (int)peer_out_count * MSG_QUEUESIZE));
	pl->estimated_load = (int)(load * 100);
	pl->active_jobs = queues_sum;
	pl->avail_threads = PEERS_MAX - peer_count;
	pl->routing_time = last_io_sum / 60;
	return 0;
}

/*
 *	peer_shut_worst_inbound()
 *	Find an inbound connection that issued the worst matching
 *	requests and inserted the most already known key, or one
 *	that did not insert or request anything at all, and kick
 *	that node off.
 */
int peer_shut_worst_inbound(void)
{
	peer_node_t *p;
	stats_t *s;
	time_t t0;
	size_t i, j;
	pid_t pid;
	double ratio, worst = 0.0;
	FUN("peer_shut_worst_inbound");

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);
	for (i = 0, j = PEERS_IN_MAX; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		s = &p->stats;
		/* give each connection 10 minutes to establish and become active */
		if (p->start + 10 * 60 >= t0) {
			/* any inbound insert or request messages at all? */
			if ((s->ins_msgs + s->req_msgs) > 0) {
				ratio = (double)(s->ins_msgs_hit + s->req_msgs_miss) /
					(double)(s->ins_msgs + s->req_msgs);
				if (ratio >= worst) {
					worst = ratio;
					j = i;
				}
			} else {
				/* not a single request in 10 minutes is certainly poor */
				j = i;
				worst = 1.0;
			}
		}
	}

	if (j >= PEERS_IN_MAX) {
		LOGS(L_PEER,L_MINOR,("found no poor inbound connection\n"));
		GLOBAL_UNLOCK();
		errno = ENOENT;
		return -1;
	}

	if (NULL == (p = g_peer->peers_in[j])) {
		LOGS(L_PEER,L_ERROR,("poor inbound connection could not be locked\n"));
		GLOBAL_UNLOCK();
		errno = ENOENT;
		return -1;
	}
	LOGS(L_PEER,L_MINOR,("worst inbound is #%u (%s) " \
		"ins/hit+req/miss ratio %d%%\n",
		(unsigned)j, p->conn.peeraddr, (int)(worst * 100.0)));
	pid = p->pid;
	GLOBAL_UNLOCK();
	kill(pid, SIGHUP);
	osd_usleep(50000);
	return 0;
}

/*
 *	peer_shut_worst_outbound()
 *	Find an outbound connection that did accept the least inserts
 *	over time and kick it off.
 */
int peer_shut_worst_outbound(void)
{
	peer_node_t *p;
	stats_t *s;
	time_t t0;
	size_t i, j;
	pid_t pid;
	double ratio, worst = 99999999.0;
	FUN("peer_shut_worst_outbound");

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);
	for (i = 0, j = PEERS_OUT_MAX; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		s = &p->stats;
		/* give each connection 10 minutes to establish and become active */
		if (p->start + 10 * 60 >= t0) {
			/* any request or insert messages at all? */
			if ((s->req_msgs + s->ins_msgs) > 0) {
				ratio = (double)(s->req_msgs + s->ins_msgs) /
					(double)(t0 - p->start);
				/* compare message ratio for this peer */
				if (ratio < worst) {
					j = i;
					worst = ratio;
				}
			} else {
				/* not a single ins/req in 10 minutes is certainly poor */
				j = i;
				worst = 0.0;
			}
		}
	}

	if (j >= PEERS_OUT_MAX) {
		LOGS(L_PEER,L_MINOR,("found no poor outbound connection\n"));
		GLOBAL_UNLOCK();
		errno = ENOENT;
		return -1;
	}

	if (NULL == (p = g_peer->peers_out[j])) {
		LOGS(L_PEER,L_ERROR,("poor outbound connection could not be locked\n"));
		GLOBAL_UNLOCK();
		errno = ENOENT;
		return -1;
	}
	LOGS(L_PEER,L_MINOR,("worst outbound is #%u (%s) " \
		"req+ins/time ratio %dmsgs/s\n",
		(unsigned)j, p->conn.peeraddr, (int)(worst * 100.0)));
	pid = p->pid;
	GLOBAL_UNLOCK();
	kill(pid, SIGHUP);
	osd_usleep(50000);
	return 0;
}

static int peer_got_fec(struct sockaddr_in *peeraddr, peer_msg_t *msg, int htl)
{
	sha1_digest_t s[1+12];
	fec_status_t st;
	size_t hashsize;
	size_t size;
	size_t i;
	size_t head;
	size_t requests;
	time_t t0;
	fec_queue_t *f;
	fec_entry_t *e;
	int rc;
	FUN("peer_got_fec");

#if	DEBUG == 0
	(void)peeraddr;
#endif

	if (0 != memcmp(KEY_VERSION, msg->data, 4)) {
		LOGS(L_PEER,L_ERROR,("invalid header data (not '%s') from %s\n",
			KEY_VERSION, sock_ntoa(peeraddr)));
		errno = EINVAL;
		return -1;
	}
	size = ntohl(*(uint32_t*)&msg->data[4]);
	hashsize = ntohl(*(uint32_t*)&msg->data[8]);

	if (hashsize != 12 * SHA1SIZE) {
		LOGS(L_PEER,L_ERROR,("hashsize:%#x != %#x from %s\n",
			(unsigned)size, 1 + 12 * SHA1SIZE + 1,
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		return -1;
	}

	if (size != 4 + 4 + 4 + 1 + hashsize + 1) {
		LOGS(L_PEER,L_ERROR,("buffsize:%#x != %#x from %s\n",
			(unsigned)size, (unsigned)(4 + 4 + 4 + 1 + hashsize + 1),
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		return -1;
	}

	if (0x55 != msg->data[size - 1]) {
		LOGS(L_PEER,L_ERROR,("trailing byte %#02x != %#02x from %s\n",
			msg->data[size - 1], 0x55,
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		return -1;
	}

	s[0] = msg->sha1;
	/* copy the 12 hashes */
	for (i = 0; i < 12; i++)
		memcpy(&s[1 + i], &msg->data[12 + 1 + i * SHA1SIZE], SHA1SIZE);
	/* hash of the 13 hashes */
	memset(&st, 0, sizeof(st));
	sha1(s, sizeof(s), &st.sha1);

	if (0 != FEC_LOCK()) {
		LOGS(L_PEER,L_ERROR,("g_peer fec lock failed (%s)\n",
			strerror(errno)));
		return -1;
	}

	t0 = time(NULL);

	if (0 != key_fec_status_get(&st)) {
		FEC_UNLOCK();
		LOGS(L_PEER,L_ERROR,("key_fec_status_get(%s) failed (%s)\n",
			sha1_hexshort(&st.sha1), strerror(errno)));
		return -1;
	}

	if (FEC_ERROR == st.status) {
		FEC_UNLOCK();
		LOGS(L_PEER,L_DEBUG,("FEC %s is a bad request\n",
			sha1_hexshort(&st.sha1)));
		errno = EINVAL;
		return -1;
	}

	if (FEC_DONE == st.status) {
		/* after five minutes allow requeueing a FEC request */
		if (st.time + 5 * 60 > t0) {
			LOGS(L_PEER,L_DEBUG,("FEC %s re-queue (last %s)\n",
				sha1_hexshort(&st.sha1), datetime_str(st.time)));
			st.status = FEC_UNUSED;
		} else {
			FEC_UNLOCK();
			LOGS(L_PEER,L_DEBUG,("FEC %s already done\n",
				sha1_hexshort(&st.sha1)));
			errno = EEXIST;
			return -1;
		}
	}

	if (FEC_PENDING == st.status) {
		FEC_UNLOCK();
		LOGS(L_PEER,L_DEBUG,("FEC %s is pending\n",
			sha1_hexshort(&st.sha1)));
		errno = EAGAIN;
		return -1;
	}

	if (htl > 1 && htl <= 5) {
		f = &g_peer->fec_queue;
		if (0 == (size = f->size)) {
			LOGS(L_PEER,L_DEBUG,("FEC queue is disabled\n"));
		} else if ((f->head + 1) % size == f->tail) {
			LOGS(L_PEER,L_DEBUG,("FEC queue is full\n"));
		} else {
			/* fill in the fec queue entry */
			head = f->head;
			e = &f->entry[head];
			memcpy(e->sha1, s, sizeof(e->sha1));
			e->start = st.time = t0;
			e->htl = htl;
			e->req = st.sha1;
			LOGS(L_PEER,L_MINOR,("FEC %s w/ HTL:%d enqueued as #%#x at %s\n",
				sha1_hexshort(&st.sha1),
				e->htl,
				(unsigned)f->head,
				datetime_str(e->start)));
			f->head = (head + 1) % size;
			st.status = FEC_PENDING;
			key_fec_status_set(&st);
		}
	}
	FEC_UNLOCK();

	if (htl > 0) {
		rc = peer_req_fec(s, 2, htl, &requests, 0);
		if (0 == rc) {
			LOGS(L_PEER,L_MINOR,("FEC %s forwarded to %u peers\n",
				sha1_hexshort(s), (unsigned)requests));
		} else {
			LOGS(L_PEER,L_DEBUG,("FEC forwarding failed (%s)\n",
				strerror(errno)));
		}
	}

	return 0;
}

/*
 *	peer_got_broadcast()
 *	Analyze a broadcast message and forward it if the HTL is greater
 *	than zero. Extract broadcast info into shared memory structure.
 */
static int peer_got_broadcast(struct sockaddr_in *peeraddr, peer_msg_t *msg, int htl_out)
{
	broadcast_msg_t *bcm;
	char *xml = NULL;
	char *name = NULL;
	size_t i, size, xmlsize;
	sha1_digest_t s;
	XML_Parser par = NULL;
	broadcast_t *bc = NULL;
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	time_t t0;
	int rc = -1;
	FUN("peer_got_broadcast");

	t0 = time(NULL);
	sha1(msg->data, msg->size, &s);

	if (0 != memcmp(&s, &msg->sha1, SHA1SIZE)) {
		LOGS(L_PEER,L_ERROR,("SHA1 mismatch (%s:%s)\n%s\n",
			sha1_hexshort(&s),
			sha1_hexshort(&msg->sha1),
			hexdump(msg->data, msg->size)));
		errno = EINVAL;
		return -1;
	}

	if (0 != memcmp(KEY_VERSION, msg->data, 4)) {
		LOGS(L_PEER,L_ERROR,("invalid broadcast (not '%s')\n",
			KEY_VERSION));
		errno = EINVAL;
		return -1;
	}

	size = ntohl(*(uint32_t*)&msg->data[4]);

	if (size < 4 + 12 + 4 + 4) {
		LOGS(L_PEER,L_ERROR,("invalid broadcast (too small)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	if (size > MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid broadcast (too big)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	xmlsize = ntohl(*(uint32_t*)&msg->data[8]);
	if (xmlsize <= 0 || xmlsize >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid broadcast (too big)\nxmlsize:%#x key:%s\n",
			(unsigned)xmlsize,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	if (0 != BCM_LOCK()) {
		LOGS(L_PEER,L_ERROR,("g_peer bcm lock failed (%s)\n",
			strerror(errno)));
		return -1;
	}

	for (i = g_peer->bcm_tail; i != g_peer->bcm_head; i = (i + 1) % BROADCAST_MAX) {
		bcm = &g_peer->bcm[i];
		if (0 == memcmp(&bcm->sha1, &msg->sha1, SHA1SIZE)) {
			/* we already have this message */
			goto bailout;
		}
	}

	rc = file_expand(&xml, &xmlsize, msg->data, size);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_expand(...,%p,%#x) failed (%s)\n",
			msg->data, (unsigned)size, strerror(errno)));
		rc = -1;
		goto bailout;
	}

	/* check that this is a XML text */
	if (0 != strncmp(xml, "<?xml", 5)) {
		LOGS(L_PEER,L_ERROR,("data does not look like an XML block\n"));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("got broadcast XML:\n%s\n", xml));

	if (NULL == (par = XML_ParserCreate(NULL))) {
		LOGS(L_PEER,L_ERROR,("XML_ParserCreate() failed\n"));
		errno = ENOMEM;
		rc = -1;
		goto bailout;
	}

	bc = (broadcast_t *)xcalloc(1, sizeof(broadcast_t));
	XML_SetElementHandler(par,
		(XML_StartElementHandler)broadcast_xml_start_element,
		(XML_EndElementHandler)broadcast_xml_end_element);
	XML_SetCharacterDataHandler(par,
		(XML_CharacterDataHandler)broadcast_xml_cdata);
	XML_SetDefaultHandler(par,
		(XML_CharacterDataHandler)broadcast_xml_default);
	XML_SetUserData(par, (void *)bc);
	if (0 == XML_Parse(par, xml, xmlsize, 1)) {
		rc = XML_GetErrorCode(par);
		LOGS(L_PEER,L_ERROR,("XML_Parse returned %d (line %d, pos %d)\n%s\n",
			rc, XML_GetCurrentLineNumber(par),
			XML_GetCurrentColumnNumber(par),
			xml));
		goto bailout;
	}

	if (0 == bc->channel.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <channel> tag in broadcast msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 == bc->sender.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <sender> tag in broadcast msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 == bc->payload.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <payload> tag in broadcast msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 != peer_get_name(peeraddr, &name)) {
		xfree(name);
	}

	LOGS(L_PEER,L_DEBUG,("Got broadcast message\n" \
		" from host:port %s (%s)\n" \
		" SHA1: %s\n" \
		" time: %x\n" \
		" channel: %s\n" \
		" sender: %s\n" \
		" payload: %s\n",
		sock_ntoa(peeraddr),
		NULL == name ? "NONAME" : name,
		hexstr(msg->sha1.digest, SHA1SIZE),
		(unsigned)t0,
		bc->channel.name,
		bc->sender.name,
		bc->payload.text));

	/* enter the broadcast into the ring buffer */
	bcm = &g_peer->bcm[g_peer->bcm_head];
	g_peer->bcm_head = (g_peer->bcm_head + 1) % BROADCAST_MAX;
	if (g_peer->bcm_head == g_peer->bcm_tail)
		g_peer->bcm_tail = (g_peer->bcm_head + 1) % BROADCAST_MAX;
	g_peer->bcm_count += 1;
	bcm->id = g_peer->bcm_count;
	bcm->sha1 = msg->sha1;
	bcm->t0 = time(NULL);
	pm_snprintf(bcm->channel, MAXCHANNEL, "%s", bc->channel.name);
	pm_snprintf(bcm->sender, MAXSENDER, "%s", bc->sender.name);
	pm_snprintf(bcm->payload, MAXPAYLOAD, "%s", bc->payload.text);

	if (0 != (rc = GLOBAL_LOCK()))
		goto bailout;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		q = &p->queue;
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		e = &q->entry[q->head];
		e->cmd = MSG_BCM;
		e->htl = htl_out;
		e->sha1 = msg->sha1;
		e->size = msg->size;
		/* size is guaranteed to be <= MSG_BUFFSIZE */
		if (e->size <= MSG_BUFFSIZE) {
			/* message data fits in buffer */
			memcpy(e->data, msg->data, msg->size);
		} else {
			die(1,"Fatal: size > MSG_BUFFSIZE\n");
		}
		QUEUE_NEXT(q);
		LOGS(L_PEER,L_DEBUG,("BCM %s to %s on queue #%d/%d\n",
			sha1_hexshort(&msg->sha1), p->conn.peeraddr,
			(int)i, (int)q->head));
	}
	GLOBAL_UNLOCK();


bailout:
	BCM_UNLOCK();
	if (0 == rc && NULL != bc && htl_out > 0)
		rc = peer_ins_broadcast(bc->channel.name, bc->sender.name, bc->payload.text, htl_out);
	xfree(name);
	xfree(xml);
	if (NULL != par) {
		XML_ParserFree(par);
		par = NULL;
	}
	xfree(bc);
	return rc;
}

/*
 *	peer_ins_broadcast()
 *	Insert a broadcast message into the queues of all outgoing connections
 *	that have some space left. The broadcast message will be forwarded
 *	to all peers until the hops to live counted down to zero.
 */
int peer_ins_broadcast(const char *channel, const char *sender, const char *payload, int htl)
{
	char *xml = NULL;
	uint8_t *buff = NULL;
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	time_t t0;
	size_t i, size, xmlsize;
	sha1_digest_t s;
	int rc = 0;
	FUN("peer_ins_broadcast");


	xmlsize = pm_asprintf(&xml,
		"<?xml version=\"1.0\" standalone='yes'?>\n" \
		"<broadcast>\n" \
		"\t<channel name='%s' />\n" \
		"\t<sender name='%s' />\n" \
		"\t<payload text='%s' />\n" \
		"</broadcast>\n",
		xml_escape(channel),
		xml_escape(sender),
		xml_escape(payload));

	if (xmlsize <= 0 || xmlsize >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! xmlsize %#x\n",
			(unsigned)xmlsize));
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("made broadcast XML:\n%s\n", xml));

	rc = file_squeeze(&buff, &size, xml, xmlsize);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_squeeze(...,%p,%#x) failed (%d)\n",
			xml, (unsigned)xmlsize, rc));
		goto bailout;
	}
	xfree(xml);

	if (size >= MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! gzsize %#x\n",
			(unsigned)size));
		rc = -1;
		goto bailout;
	}

	/* make the SHA1 of the buffer */
	sha1(buff, size, &s);

	if (0 != (rc = GLOBAL_LOCK()))
		goto bailout;

	t0 = time(NULL);

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		q = &p->queue;
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		e = &q->entry[q->head];
		e->cmd = MSG_BCM;
		e->htl = htl;
		e->sha1 = s;
		e->size = size;
		/* size is guaranteed to be <= MSG_BUFFSIZE */
		if (e->size <= MSG_BUFFSIZE) {
			/* message data fits in buffer */
			memcpy(e->data, buff, size);
		} else {
			die(1,"Fatal: size > MSG_BUFFSIZE\n");
		}
		QUEUE_NEXT(q);
		LOGS(L_PEER,L_DEBUG,("BCM %s to %s on queue #%d/%d\n",
			sha1_hexshort(&s), p->conn.peeraddr,
			(int)i, (int)q->head));
	}
	GLOBAL_UNLOCK();

bailout:
	xfree(buff);
	xfree(xml);
	return rc;
}

/*
 *	peer_get_broadcast()
 *	Get a broadcast message with identifier 'id' from the list of
 *	recently received broadcasts. If 'id' is zero, return the min
 *	and max available IDs, otherwise fill in the callers buffers
 *	with the message data.
 */
int peer_get_broadcast(int id, int *pmin, int *pmax,
	sha1_digest_t *pd, time_t *pt,
	char *channel, char *sender, char *payload)
{
	broadcast_msg_t *bcm;
	size_t i;
	FUN("peer_get_broadcast");

	if (0 != BCM_LOCK()) {
		LOGS(L_PEER,L_ERROR,("g_peer bcm lock failed (%s)\n",
			strerror(errno)));
		errno = EAGAIN;
		return -1;
	}
	if (0 == id) {
		*pmin = 0x7fffffff;
		*pmax = 0;
		for (i = g_peer->bcm_tail; i != g_peer->bcm_head; i = (i + 1) % BROADCAST_MAX) {
			bcm = &g_peer->bcm[i];
			if (bcm->id < *pmin)
				*pmin = bcm->id;
			if (bcm->id > *pmax)
				*pmax = bcm->id;
		}
		if (*pmin > *pmax)
			*pmin = *pmax = 0;
		BCM_UNLOCK();
		return 0;
	}
	for (i = g_peer->bcm_tail; i != g_peer->bcm_head; i = (i + 1) % BROADCAST_MAX) {
		bcm = &g_peer->bcm[i];
		if (id == bcm->id) {
			if (NULL != pd)
				*pd = bcm->sha1;
			if (NULL != pt)
				*pt = bcm->t0;
			if (NULL != channel)
				pm_snprintf(channel, MAXCHANNEL, "%s", bcm->channel);
			if (NULL != sender)
				pm_snprintf(sender, MAXSENDER, "%s", bcm->sender);
			if (NULL != payload)
				pm_snprintf(payload, MAXPAYLOAD, "%s", bcm->payload);
			BCM_UNLOCK();
			return 0;
		}
	}
	BCM_UNLOCK();
	errno = ENOENT;
	return -1;
}

/*
 *	peer_got_fpr()
 *	Check to see if the message that was inserted from a peer with HTL 0
 *	does contain a squeezed XML text with a <store> tag containing a
 *	fingerprint and total and current sizes.
 */
static int peer_got_fpr(struct sockaddr_in *peeraddr, peer_msg_t *msg)
{
	char *xml = NULL;
	char *name = NULL;
	size_t size, xmlsize, crypto_module;
	sha1_digest_t s;
	ek5_digest_t ek;
	XML_Parser par = NULL;
	p2p_info_t *pi = NULL;
	int rc = -1;
	FUN("peer_got_fpr");

	sha1(msg->data, msg->size, &s);

	if (0 != memcmp(&s, &msg->sha1, SHA1SIZE)) {
		LOGS(L_PEER,L_ERROR,("SHA1 mismatch (%s:%s)\n%s\n",
			sha1_hexshort(&s),
			sha1_hexshort(&msg->sha1),
			hexdump(msg->data, msg->size)));
		errno = EINVAL;
		return -1;
	}

	if (0 != memcmp(KEY_VERSION, msg->data, 4)) {
		LOGS(L_PEER,L_ERROR,("invalid fingerprint (not '%s')\n",
			KEY_VERSION));
		errno = EINVAL;
		return -1;
	}

	size = ntohl(*(uint32_t*)&msg->data[4]);

	if (size < 4 + 12 + 4 + 4) {
		LOGS(L_PEER,L_ERROR,("invalid fingerprint (too small)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	if (size > MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid fingerprint (too big)\nsize:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	xmlsize = ntohl(*(uint32_t*)&msg->data[8]);
	if (xmlsize <= 0 || xmlsize >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid fingerprint (too big)\nxmlsize:%#x key:%s\n",
			(unsigned)xmlsize,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	/* make the decryption key from the ENTROPY string constant */
	ek5(EK5_ENTROPY, strlen(EK5_ENTROPY), &ek);

	/* and decrypt with this key */
	key_crypt(ek.digest, msg->data, msg->size, KEY_CRYPT_GZ);

	rc = file_expand(&xml, &xmlsize, msg->data, size);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_expand(...,%p,%#x) failed (%s)\n",
			msg->data, (unsigned)size, strerror(errno)));
		rc = -1;
		goto bailout;
	}

	/* check that this is a XML text */
	if (0 != strncmp(xml, "<?xml", 5)) {
		LOGS(L_PEER,L_ERROR,("data does not look like an XML block\n"));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("got fingerprint XML:\n%s\n", xml));

	if (NULL == (par = XML_ParserCreate(NULL))) {
		LOGS(L_PEER,L_ERROR,("XML_ParserCreate() failed\n"));
		errno = ENOMEM;
		rc = -1;
		goto bailout;
	}

	pi = (p2p_info_t *)xcalloc(1, sizeof(p2p_info_t));
	XML_SetElementHandler(par,
		(XML_StartElementHandler)p2p_info_xml_start_element,
		(XML_EndElementHandler)p2p_info_xml_end_element);
	XML_SetCharacterDataHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_cdata);
	XML_SetDefaultHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_default);
	XML_SetUserData(par, (void *)pi);
	if (0 == XML_Parse(par, xml, xmlsize, 1)) {
		rc = XML_GetErrorCode(par);
		LOGS(L_PEER,L_ERROR,("XML_Parse returned %d (line %d, pos %d)\n%s\n",
			rc, XML_GetCurrentLineNumber(par),
			XML_GetCurrentColumnNumber(par),
			xml));
		goto bailout;
	}

	if (0 == pi->peer.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <peer ...> tag in fingerprint msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 == pi->store.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <store ...> tag in fingerprint msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 == pi->node.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <node ...> tag in fingerprint msg from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 != peer_get_name(peeraddr, &name)) {
		xfree(name);
	}

	LOGS(L_PEER,L_MINOR,("Got fingerprint info\n" \
		" from host:port %s (%s)\n" \
		" she claims to be %s\n" \
		" node info %s-%d.%d.%d-%d\n" \
		" store size %s of %s\n" \
		" fingerprint [%s]\n" \
		" preferred crypto module '%s'\n",
		sock_ntoa(peeraddr),
		NULL == name ? "NONAME" : name,
		sock_ntoa(&pi->peer.addr),
		pi->node.type, pi->node.major,
		MINOR_LEVEL(pi->node.minor),
		MINOR_STEP(pi->node.minor),
		pi->node.build,
		round_KMG(pi->store.current),
		round_KMG(pi->store.size),
		hexstr(pi->store.fingerprint, 16),
		(char *)(pi->crypto.found_tag ? pi->crypto.module : "unknown")));

	if (peeraddr->sin_addr.s_addr != pi->peer.addr.sin_addr.s_addr) {
		/* Add the wrong local IP to the failure list */
		peer_add_failure(&pi->peer.addr);
	}

	peer_set_node(peeraddr,
		pi->node.type, pi->node.major, pi->node.minor, pi->node.build);
	peer_set_fingerprint(peeraddr,
		pi->store.fingerprint);
	peer_set_size_current(peeraddr,
		pi->store.size, pi->store.current);
	if (0 != pi->crypto.found_tag) {
		crypto_module = crypto_module_indx(pi->crypto.module);
		peer_set_crypto_module_in(peeraddr, crypto_module);
	}

bailout:
	xfree(name);
	xfree(xml);
	if (NULL != par) {
		XML_ParserFree(par);
		par = NULL;
	}
	xfree(pi);
	return rc;
}

/*
 *	peer_ins_fpr()
 *	Insert our fingerprint into the queues of all outgoing connections
 *	that have some space left. The insert message has a HTL of zero and
 *	is detected at the receiver's end.
 */
int peer_ins_fpr(struct sockaddr_in *peeraddr, uint8_t fingerprint[16])
{
	char *xml = NULL;
	uint8_t *buff = NULL;
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	time_t t0;
	size_t i, size, xmlsize;
	sha1_digest_t s;
	ek5_digest_t ek;
	int rc = 0;
	FUN("peer_ins_fpr");

	xmlsize = pm_asprintf(&xml,
		"<?xml version=\"1.0\" standalone='yes'?>\n" \
		"<p2p>\n" \
		"\t<peer address='%s' hostname='%s' port='%d' />\n" \
		"\t<node type='%s' major='%s' minor='%s' build='%s' />\n" \
		"\t<store fingerprint='%s' size='%llx' current='%llx' />\n" \
		"\t<crypto module='%s' />\n" \
		"</p2p>\n",
		sock_ntoa(peeraddr),
		inet_ntoa(peeraddr->sin_addr),
		ntohs(peeraddr->sin_port),
		PROGNAME, NODE_VER_MAJ, NODE_VER_MIN, NODE_BUILD,
		hexstr(fingerprint, 16), g_store->storesize, g_store->currsize,
		g_conf->crypto_default);

	if (xmlsize <= 0 || xmlsize >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! xmlsize %#x\n",
			(unsigned)xmlsize));
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("made fingerprint XML:\n%s\n", xml));

	rc = file_squeeze(&buff, &size, xml, xmlsize);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_squeeze(...,%p,%#x) failed (%d)\n",
			xml, (unsigned)xmlsize, rc));
		goto bailout;
	}
	xfree(xml);

	if (size >= MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! size %#x\n",
			(unsigned)size));
		rc = -1;
		goto bailout;
	}

	/* make the key from the ENTROPY string constant */
	ek5(EK5_ENTROPY, strlen(EK5_ENTROPY), &ek);

	/* and encrypt with the EK5 of this key */
	key_crypt(ek.digest, buff, size, KEY_CRYPT_GZ);

	/* make the SHA1 of the encrypted buffer */
	sha1(buff, size, &s);

	if (0 != (rc = GLOBAL_LOCK()))
		goto bailout;

	t0 = time(NULL);

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		q = &p->queue;
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		e = &q->entry[q->head];
		e->cmd = MSG_FPR;
		e->htl = 0;
		e->sha1 = s;
		e->size = size;
		if (e->size <= MSG_BUFFSIZE) {
			/* message data fits in buffer */
			memcpy(e->data, buff, size);
		} else {
			die(1,"Fatal: size > MSG_BUFFSIZE\n");
		}
		QUEUE_NEXT(q);
		LOGS(L_PEER,L_DEBUG,("FPR %s to %s on queue #%d/%d\n",
			sha1_hexshort(&s), p->conn.peeraddr,
			(int)i, (int)q->head));
	}
	GLOBAL_UNLOCK();

bailout:
	xfree(buff);
	xfree(xml);
	return rc;
}

/*
 *	peer_ins_ann()
 *	Announce a node.
 */
int peer_ins_ann(struct sockaddr_in *peeraddr, size_t crypto_module, int htl)
{
	char *xml = NULL;
	uint8_t *buff = NULL;
	char module[32];
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	announce_t *a;
	time_t t0;
	size_t i, size, xmlsize;
	sha1_digest_t s;
	ek5_digest_t ek;
	int rc = 0;
	long addr;
	FUN("peer_ins_ann");

	pm_snprintf(module, sizeof(module), "%s",
		crypto_module_name(crypto_module));

	addr = ntohl(peeraddr->sin_addr.s_addr);
	if (is_local_ip(addr, 0)) {
		LOGS(L_PEER,L_DEBUG,("announce locally (local IP)\n"));
		htl = 0;
	} else if (0 == g_peer->peercnt_out) {
		LOGS(L_PEER,L_DEBUG,("announce locally (no outbound connections)\n"));
		htl = 0;
	}

	xmlsize = pm_asprintf(&xml,
		"<?xml version=\"1.0\" standalone='yes'?>\n" \
		"<p2p>\n" \
		"\t<peer address='%s' hostname='%s' port='%d' />\n" \
		"\t<crypto module='%s' />\n" \
		"</p2p>\n",
		sock_ntoa(peeraddr),
		inet_ntoa(peeraddr->sin_addr),
		ntohs(peeraddr->sin_port),
		module);

	if (xmlsize <= 0 || xmlsize >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! xmlsize %#x\n",
			(unsigned)xmlsize));
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("made announcement XML:\n%s\n", xml));

	rc = file_squeeze(&buff, &size, xml, xmlsize);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_squeeze(...,%p,%#x) failed (%d)\n",
			xml, (unsigned)xmlsize, rc));
		goto bailout;
	}
	xfree(xml);

	if (size <= 0 || size > MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("Houston, we have a problem! gzsize %#x\n",
			(unsigned)size));
		rc = -1;
		goto bailout;
	}

	/* make the key from the ENTROPY string constant */
	ek5(EK5_ENTROPY, strlen(EK5_ENTROPY), &ek);

	/* and encrypt with the EK5 of this key */
	key_crypt(ek.digest, buff, size, KEY_CRYPT_GZ);

	/* make the SHA1 of the encrypted buffer */
	sha1(buff, size, &s);

	t0 = time(NULL);

	if (0 != (rc = GLOBAL_LOCK()))
		goto bailout;

	/* search the list if we already have this entry (disregard module) */
	for (i = 0; i < ANNOUNCE_MAX; i++) {
		a = &g_peer->ann[i];
		if (0 == memcmp(&a->addr, peeraddr, sizeof(a->addr)))
			break;
	}

	/* make a new entry if we did not find it, or in case it is too old */
	if (i >= ANNOUNCE_MAX || a->last + 30 * 60 < t0) {
		if (i >= ANNOUNCE_MAX) {
			a = &g_peer->ann[g_peer->announce_in];
			g_peer->announce_in = (g_peer->announce_in + 1) % ANNOUNCE_MAX;
			a->addr = *peeraddr;
			pm_snprintf(a->module, sizeof(a->module), "%s", module);
		}
		a->last = t0;

		/* forward only if htl is greater than zero */
		if (htl > 0) {

			for (i = 0; i < g_peer->peercnt_out; i++) {
				if (NULL == (p = g_peer->peers_out[i])) {
					/* unused slot */
					continue;
				}
				if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
					/* not connected */
					continue;
				}
				q = &p->queue;
				if (0 != QUEUE_FULL(q)) {
					/* outbound queue is full */
					continue;
				}

				e = &q->entry[q->head];
				e->cmd = MSG_ANN;
				e->htl = htl;
				e->sha1 = s;
				e->size = size;
				if (e->size <= MSG_BUFFSIZE) {
					/* data fits in buffer */
					memcpy(e->data, buff, size);
				} else {
					die(1, "Fatal: size > MSG_BUFFSIZE\n");
				}
				QUEUE_NEXT(q);
				LOGS(L_PEER,L_DEBUG,("ANN %s to %s on queue #%d/%d\n",
					sha1_hexshort(&s), p->conn.peeraddr,
					(int)i, (int)q->head));
			}
		}
	}
	GLOBAL_UNLOCK();

bailout:
	xfree(buff);
	xfree(xml);
	return rc;
}
 
/*
 *	peer_got_ann()
 *	Check to see if the message that was inserted from a peer with HTL 0
 *	does contain a squeezed XML text with a <peer> tag containing a
 *	hostname and port, as well as a <crypto> tag containing a module name.
 */
static int peer_got_ann(struct sockaddr_in *peeraddr, peer_msg_t *msg, int htl)
{
	char *xml = NULL;
	char *name = NULL;
	size_t size, xmlsize, crypto_module;
	sha1_digest_t s;
	ek5_digest_t ek;
	XML_Parser par = NULL;
	p2p_info_t *pi = NULL;
	int rc = -1;
	FUN("peer_got_ann");

	sha1(msg->data, msg->size, &s);

	if (0 != memcmp(&s, &msg->sha1, SHA1SIZE)) {
		LOGS(L_PEER,L_ERROR,("SHA1 mismatch (%s:%s)\n%s\n",
			sha1_hexshort(&s),
			sha1_hexshort(&msg->sha1),
			hexdump(msg->data, msg->size)));
		errno = EINVAL;
		return -1;
	}

	if (0 != memcmp(KEY_VERSION, msg->data, 4)) {
		LOGS(L_PEER,L_ERROR,("invalid announcement (not '%s')\n",
			KEY_VERSION));
		errno = EINVAL;
		return -1;
	}

	size = ntohl(*(uint32_t*)&msg->data[4]);

	if (size < 4 + 12 + 4 + 4) {
		LOGS(L_PEER,L_ERROR,("invalid announcement (too small)\n" \
			"size:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	if (size >= MSG_BUFFSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid announcement (too big)\n" \
			"size:%#x key:%s\n",
			(unsigned)size,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	xmlsize = ntohl(*(uint32_t*)&msg->data[8]);
	if (xmlsize <= 0 || size >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("invalid announcement (bad xmlsize)\n" \
			"xmlsize:%#x key:%s\n",
			(unsigned)xmlsize,
			sha1_hexshort(&msg->sha1)));
		errno = EINVAL;
		return -1;
	}

	/* make the decryption key from the ENTROPY string constant */
	ek5(EK5_ENTROPY, strlen(EK5_ENTROPY), &ek);

	/* and decrypt with this key */
	key_crypt(ek.digest, msg->data, msg->size, KEY_CRYPT_GZ);

	rc = file_expand(&xml, &xmlsize, msg->data, size);
	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("file_expand(...,%p,%#x) failed (%s)\n",
			msg->data, (unsigned)size, strerror(errno)));
		rc = -1;
		goto bailout;
	}

	/* check that this is a XML text */
	if (0 != strncmp(xml, "<?xml", 5)) {
		LOGS(L_PEER,L_ERROR,("data does not look like an XML block\n"));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (NULL == (par = XML_ParserCreate(NULL))) {
		LOGS(L_PEER,L_ERROR,("XML_ParserCreate() failed\n"));
		errno = ENOMEM;
		rc = -1;
		goto bailout;
	}

	pi = (p2p_info_t *)xcalloc(1, sizeof(p2p_info_t));
	XML_SetElementHandler(par,
		(XML_StartElementHandler)p2p_info_xml_start_element,
		(XML_EndElementHandler)p2p_info_xml_end_element);
	XML_SetCharacterDataHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_cdata);
	XML_SetDefaultHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_default);
	XML_SetUserData(par, (void *)pi);
	if (0 == XML_Parse(par, xml, xmlsize, 1)) {
		rc = XML_GetErrorCode(par);
		LOGS(L_PEER,L_ERROR,("XML_Parse returned %d (line %d, pos %d)\n%s\n",
			rc, XML_GetCurrentLineNumber(par),
			XML_GetCurrentColumnNumber(par),
			xml));
		goto bailout;
	}

	if (0 == pi->peer.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <peer ...> tag in announcement from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (0 == pi->crypto.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <crypto ...> tag in announcement from %s\n",
			sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	crypto_module = crypto_module_indx(pi->crypto.module);
	if ((size_t)-1 == crypto_module) {
		LOGS(L_PEER,L_ERROR,("Unknown crypto module '%s' from %s\n",
			pi->crypto.module, sock_ntoa(peeraddr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	if (-1 == crypto_module_supp(crypto_module)) {
		LOGS(L_PEER,L_MINOR,("Rejected crypto module '%s' from %s\n",
			pi->crypto.module,
			sock_ntoa(peeraddr)));
		crypto_module = crypto_module_indx(g_conf->crypto_default);
	}

	if (pi->peer.addr.sin_addr.s_addr == (uint32_t)-1 ||
		pi->peer.addr.sin_addr.s_addr == (uint32_t)0) {
		LOGS(L_PEER,L_ERROR,("peer address (%s) is invalid\n",
			sock_ntoa(&pi->peer.addr)));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}
	if (ntohs(pi->peer.addr.sin_port) < 1024) {
		LOGS(L_PEER,L_ERROR,("peer port (%s:%d) is not acceptable\n",
			 sock_ntoa(&pi->peer.addr),
			 (int)(ntohs(pi->peer.addr.sin_port))));
		goto bailout;
	}

	if (0 != peer_get_name(peeraddr, &name)) {
		xfree(name);
	}

	LOGS(L_PEER,L_DEBUG,("Got announcement\n" \
		" from host:port %s (%s)\n" \
		" she announces %s/%s [%s]\n",
		sock_ntoa(peeraddr), NULL == name ? "NONAME" : name,
		sock_ntoa(&pi->peer.addr), pi->crypto.module,
		crypto_module_name(crypto_module)));

	if (0 != peer_blacklist_check(&pi->peer.addr)) {
		LOGS(L_PEER,L_ERROR,("peer announcement rejected\n"));
		goto bailout;
	}

	/* forward the announcement */
	peer_ins_ann(&pi->peer.addr, crypto_module, htl);

bailout:
	xfree(name);
	xfree(xml);
	if (NULL != par) {
		XML_ParserFree(par);
		par = NULL;
	}
	xfree(pi);
	return rc;
}

/*
 *	peer_node_search()
 *	Find a new peer, which is not yet in the list of connected peers
 *	and return the address (or hostname) and port number.
 */
int peer_node_search(struct sockaddr_in *peer, size_t *crypto_module)
{
	announce_t ann;
	size_t i, j, module;
	int rc = 0;
	FUN("peer_node_search");

	j = g_peer->announce_out;

	for (i = 0; i < ANNOUNCE_MAX; i++) {
		/* get the next announcement */
		if (0 != (rc = GLOBAL_LOCK()))
			return -1;
		j = (j + 1) % ANNOUNCE_MAX;
		ann = g_peer->ann[j];
		GLOBAL_UNLOCK();

		if (0 == ann.addr.sin_port || '\0' == ann.module[0]) {
			/* skip empty entry */
			continue;
		}

		if (0 == peer_connected(&ann.addr)) {
			LOGS(L_PEER,L_DEBUGX,("found %s -- already connected\n",
				sock_ntoa(&ann.addr)));
			continue;	/* skip if we're connected to that peer */
		}

		if (0 == peer_connecting(&ann.addr)) {
			LOGS(L_PEER,L_DEBUGX,("found %s -- already connecting\n",
				sock_ntoa(&ann.addr)));
			continue;	/* skip if we're connecting to that peer */
		}

		if (0 == peer_blocked(&ann.addr)) {
			LOGS(L_PEER,L_DEBUG,("found %s -- blocked\n",
				sock_ntoa(&ann.addr)));
			continue;	/* skip if this ip/port is blocked */
		}

		LOGS(L_PEER,L_MINOR,("found %s/%s announcement\n",
			sock_ntoa(&ann.addr), ann.module));

		/* check if we support this crypto module */
		module = crypto_module_indx(ann.module);
		if ((size_t)-1 == module) {
			LOGS(L_PEER,L_ERROR,("found %s -- invalid module '%s'\n",
				sock_ntoa(&ann.addr), ann.module));
			/* fall back to our default module */
			pm_snprintf(ann.module, sizeof(ann.module), "%s",
				g_conf->crypto_default);
			module = crypto_module_indx(ann.module);
		}

		if (-1 == crypto_module_supp(module)) {
			LOGS(L_PEER,L_MINOR,("found %s -- rejected module '%s'\n",
				sock_ntoa(&ann.addr), ann.module));
			/* fall back to our default module */
			pm_snprintf(ann.module, sizeof(ann.module), "%s",
				g_conf->crypto_default);
			module = crypto_module_indx(ann.module);
		}

		/* set the return values */
		*peer = ann.addr;
		*crypto_module = module;

		/* start searching after this index next time */
		g_peer->announce_out = j;
		return 0;
	}

	errno = ENOENT;
	return -1;
}

/*
 *	peer_get_name()
 *	Get the name of any peer with address 'addr' (which is non empty)
 */
static int peer_get_name(struct sockaddr_in *addr, char **pname)
{
	char *name = NULL;
	peer_node_t *p;
	size_t i;
	FUN("peer_get_name");

	*pname = NULL;

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr) {
			/* not the node we want */
			continue;
		}
		if (strlen(p->conn.peername) > 0) {
			name = xstrdup(p->conn.peername);
			*pname = name;
			GLOBAL_UNLOCK();
			return 0;
		}
	}
	GLOBAL_UNLOCK();

	errno = ENOENT;
	return -1;
}

/*
 *	peer_set_name()
 *	Set the name of all peers with address 'addr' to 'name'
 */
static int peer_set_name(struct sockaddr_in *addr, const char *peername)
{
	char *name = NULL, *colon;
	peer_node_t *p;
	size_t i;
	FUN("peer_set_name");

	if (0 != GLOBAL_LOCK())
		return -1;

	name = xstrdup(peername);
	/* strip off a port number part */
	colon = strchr(name, ':');
	if (NULL != colon)
		*colon = '\0';

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr) {
			/* not the node we want */
			continue;
		}
		pm_snprintf(p->conn.peername, sizeof(p->conn.peername), "%s:%d",
			name, ntohs(p->conn.address.sin_port));
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s name to '%s'\n",
			(int)i, p->conn.peeraddr, name));
	}
	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr) {
			/* not the node we want */
			continue;
		}
		pm_snprintf(p->conn.peername, sizeof(p->conn.peername), "%s:%d",
			name, ntohs(p->conn.address.sin_port));
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s name to '%s'\n",
			(int)i, p->conn.peeraddr, name));
	}
	GLOBAL_UNLOCK();

	xfree(name);
	return 0;
}

/*
 *	peer_set_fingerprint()
 *	Set the fingerprint of all peers with address 'addr' to 'fingerprint'
 *	Side effect: update g_peer->routing[] to the average key routing as
 *	our node sees it.
 */
static int peer_set_fingerprint(struct sockaddr_in *addr, uint8_t *fingerprint)
{
	peer_node_t *p;
	size_t sum[16], cnt[16];
	size_t i, j;
	FUN("peer_set_fingerprint");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr ||
			p->conn.address.sin_port != addr->sin_port) {
			/* not the node we want */
			continue;
		}
		memcpy(p->fingerprint, fingerprint, sizeof(p->fingerprint));
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s fingerprint to [%s]\n",
			(int)i, p->conn.peeraddr, hexstr(fingerprint, 16)));
	}
	memset(sum, 0, sizeof(sum));
	memset(cnt, 0, sizeof(cnt));
	g_peer->routing_cnt = 0;
	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->storesize <= 0 || p->currsize <= 0) {
			/* no store size info from peer */
			continue;
		}
		/* add fingerprint to sum/cnt arrays */
		for (j = 0; j < 16; j++) {
			sum[j] += p->fingerprint[j];
			cnt[j] += 1;
		}
		g_peer->routing_cnt += 1;
	}
	/* update the g_peer->routing[] array */
	for (j = 0; j < 16; j++) {
		if (cnt[j] > 0) {
			g_peer->routing[j] = sum[j] / cnt[j];
		} else {
			g_peer->routing[j] = 0;
		}
	}
	GLOBAL_UNLOCK();

	return 0;
}

/*
 *	peer_set_crypto_module_in()
 *	Set the crypto_module_in of all peers with address 'addr' to 'module'
 */
static int peer_set_crypto_module_in(struct sockaddr_in *addr, size_t module)
{
	peer_node_t *p;
	size_t i;
	FUN("peer_set_crypto_module");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr) {
			/* not the node we want */
			continue;
		}
		p->crypto_module_in = module;
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s crypto_module_in to %s\n",
			(int)i, p->conn.peeraddr, crypto_module_name(module)));
	}
	GLOBAL_UNLOCK();

	return 0;
}

/*
 *	peer_set_size_current()
 *	Set the fingerprint of all peers with address 'addr' to 'fingerprint'
 */
static int peer_set_size_current(struct sockaddr_in *addr,
	uint64_t storesize, uint64_t currsize)
{
	peer_node_t *p;
	size_t i;
	FUN("peer_set_size_current");

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr ||
			p->conn.address.sin_port != addr->sin_port) {
			/* not the node we want */
			continue;
		}

		p->storesize = storesize;
		p->currsize = currsize;
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s store size to %s of %s\n",
			(int)i, p->conn.peeraddr,
			round_KMG(currsize),
			round_KMG(storesize)));
	}
	GLOBAL_UNLOCK();

	return 0;
}

/*
 *	peer_set_node()
 *	Set the node type and version/build numbers of all peers
 *	with address 'addr' to 'type', 'major', 'minor' and 'build'
 */
static int peer_set_node(struct sockaddr_in *addr,
	const char *type, uint32_t major, uint32_t minor, uint32_t build)
{
	peer_node_t *p;
	size_t i;
	FUN("peer_set_node");

	if (0 != GLOBAL_LOCK())
		return -1;

	/* search for the peer connection(s) and shut down their sockets */
	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr ||
			p->conn.address.sin_port != addr->sin_port) {
			/* not the node we want */
			continue;
		}
		pm_snprintf(p->type, sizeof(p->type), "%s", type);
		p->major = major;
		p->minor = minor;
		p->build = build;
		if (build > g_conf->latest_build) {
			g_conf->latest_major = major;
			g_conf->latest_minor = minor;
			g_conf->latest_build = build;
		}
		LOGS(L_PEER,L_DEBUG,("setting peer #%d %s node to %s-%d.%d.%d-%d\n",
			(int)i, p->conn.peeraddr, type,
			major, minor/65536, minor%65536, build));
	}
	GLOBAL_UNLOCK();

	return 0;
}

/*
 *	peer_set_cancer()
 *	Set a node's cancer value to the passed argument. Higher values
 *	will cause slower delivery of output messages.
 */
static int peer_set_cancer(struct sockaddr_in *addr, int cancer)
{
	peer_node_t *p;
	size_t i;
	FUN("peer_set_cancer");

	if (0 != GLOBAL_LOCK())
		return -1;

	/* search for the peer connection(s) and set their cancer value */
	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != addr->sin_addr.s_addr ||
			p->conn.address.sin_port != addr->sin_port) {
			/* not the node we want */
			continue;
		}
		p->cancer = cancer;
		LOGS(L_PEER,L_MINOR,("setting peer #%d %s cancer count to %d\n",
			(int)i, p->conn.peeraddr, cancer));
	}
	GLOBAL_UNLOCK();

	return 0;
}
/*
 *	peer_node_maintenance()
 *	Announce all our outgoing connections. Kick out connections where the
 *	queue is stuck and that exceed a timeout of 10 minutes.
 */
int peer_node_maintenance(void)
{
	pid_t kill_pids[PEERS_MAX];
	size_t kill_count;
	struct sockaddr_in addr[PEERS_OUT_MAX];
	size_t module[PEERS_OUT_MAX];
	size_t peer_count;
	peer_node_t *p;
	struct hostent *host;
	struct sockaddr_in node_new;
	size_t i;
	time_t t0;
	int htl = ANNOUNCE_HTL;
	int diff, newip, number, rc = 0;
	FUN("peer_node_maintenance");

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
	}

	LOGS(L_PEER,L_NORMAL,("*** running node maintenance\n"));

	/* check the global routing if we have enough peer node routing infos */
	if (g_peer->routing_cnt >= PEERS_OUT_MAX/4) {
		size_t min, max, i;
		for (i = 0, min = 0; i < 16; i++) {
			if (g_peer->routing[i] < g_peer->routing[min])
				min = i;
		}
		/* see if we have a 'hole' in the global (average) routing */
		if (g_peer->routing[min] < 0x38) {
			LOGS(L_PEER,L_NORMAL,("global routing:%s min:%x/%02x\n",
				hexstr(g_peer->routing, 16),
				(unsigned)min, g_peer->routing[min]));
			/* see if we contribute to the unbalanced routing */
			if (g_store->destination[min] < 0xf0) {
				uint8_t tmp;
				/* search in our destination peaks the most common route */
				for (i = 0, max = (size_t)-1; i < 16; i++) {
					if (g_store->destination[i] < 0xf0)
						continue;
					if ((size_t)-1 == max ||
						g_peer->routing[i] >= g_peer->routing[max])
						max = i;
				}
				LOGS(L_PEER,L_NORMAL,("* ROUTING %x/%02x <-> %x/%02x *\n",
					(unsigned)min, g_store->destination[min],
					(unsigned)max, g_store->destination[max]));
				tmp = g_store->destination[min];
				g_store->destination[min] = g_store->destination[max];
				g_store->destination[max] = tmp;
				/* adjust our fingerprint with the changed destination[] */
				store_upd_fingerprint();
			}
		}
	}

	/* find out about our own address */
	memset(&node_new, 0, sizeof(node_new));
	node_new.sin_family = AF_INET;
	node_new.sin_addr.s_addr = inet_addr(g_conf->nodename);
	node_new.sin_port = htons(g_conf->nodeport);
	if (node_new.sin_addr.s_addr == (uint32_t)-1) {
		/* inet_addr failed, probably a hostname */
		host = gethostbyname(g_conf->nodename);
		if (host == NULL) {
			LOGS(L_PEER,L_DEBUG,("gethostbyname(%s) failed (%s)\n",
				g_conf->nodename, hstrerror(h_errno)));
			/* we can't resolve a new address, so we must keep the old one */
			node_new.sin_addr.s_addr =
				g_conf->node.sin_addr.s_addr;
		} else {
			node_new.sin_addr.s_addr =
				((struct in_addr *) *host->h_addr_list)->s_addr;
		}
	}

	t0 = time(NULL);
	newip = 0;
	peer_count = 0;

	/* check if we got a new IP address */
	if (node_new.sin_addr.s_addr != g_conf->node.sin_addr.s_addr) {
		LOGS(L_PEER,L_NORMAL,("***** NEW IP ADDRESS %s *****\n",
			sock_ntoa(&node_new)));
		newip = 1;
		/* reload seed nodes so that we have something to connect */
		peer_seednodes(&number);
		LOGS(L_PEER,L_NORMAL,("%d seed nodes found\n", number));
		/* copy the new address over */
		g_conf->node = node_new;
		/* announce the nodes with HTL 0, since our connections are dead */
		htl = 0;
	}
	
	if (g_conf->node.sin_addr.s_addr != (uint32_t)-1) {
		/* announce our own node */
		addr[peer_count] = g_conf->node;
		module[peer_count] = crypto_module_indx(g_conf->crypto_default);
		LOGS(L_PEER,L_MINOR,("*** announcing our node %s/%s\n",
			sock_ntoa(&addr[peer_count]),
			crypto_module_name(module[peer_count])));
		peer_count++;
	}

	LOGS(L_PEER,L_MINOR,("*** spreading our fingerprint [%s]\n",
		hexstr(g_store->fingerprint, 16)));
	if (0 != peer_ins_fpr(&g_conf->node, g_store->fingerprint)) {
		LOGS(L_PEER,L_ERROR,("peer_ins_fpr(%s/%s,[%s]) failed (%s)\n",
			sock_ntoa(&g_conf->node),
			g_conf->crypto_default,
			hexstr(g_store->fingerprint, 16),
			strerror(errno)));
	}

	kill_count = 0;

	if (0 != GLOBAL_LOCK())
		return -1;

	for (i = 0; i < g_peer->peercnt_in; i++) {
		if (NULL == (p = g_peer->peers_in[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		diff = (0 != p->conn.last_io && t0 > p->conn.last_io) ?
			t0 - p->conn.last_io : 0;
		if (0 == newip && diff < 10 * 60) {
			continue;
		}
		LOGS(L_PEER,L_ERROR,("shut inbound peer {%d} %s (quiet for %d s)\n",
			(int)p->pid, p->conn.peeraddr, diff));
		/* save pid in the list of processes to kill */
		kill_pids[kill_count++] = p->pid;
	}

	for (i = 0; i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		diff = (0 != p->conn.last_io && t0 > p->conn.last_io) ?
			t0 - p->conn.last_io : 0;
		if (0 != newip || diff >= 10 * 60) {
			LOGS(L_PEER,L_ERROR,("shut outbound peer {%d} %s (stuck for %ds)\n",
				(int)p->pid, p->conn.peeraddr, diff));
			/* save pid in the list of processes to kill */
			kill_pids[kill_count++] = p->pid;
			continue;
		}

		if ('\0' == p->type[0]) {
			LOGS(L_PEER,L_MINOR,("*** skipping node #%u %s\n",
				(unsigned)i, p->conn.peeraddr));
			continue;
		}
		/* announce outgoing connections that we know about */
		addr[peer_count] = p->conn.address;
		module[peer_count] = p->crypto_module_in;
		LOGS(L_PEER,L_MINOR,("*** announcing node %s/%s\n",
			sock_ntoa(&addr[peer_count]),
			crypto_module_name(module[peer_count])));
		peer_count++;
	}
	GLOBAL_UNLOCK();

	/* kill the processes that timed out */
	for (i = 0; i < kill_count; i++) {
		rc = kill(kill_pids[i], SIGHUP);
		if (0 != rc) {
			LOGS(L_PEER,L_ERROR,("kill(%u,SIGHUP) failed (%s)\n",
				(unsigned)kill_pids[i],
				strerror(errno)));
		}
	}

	/* announce the peer nodes that were active */
	for (i = 0; i < peer_count; i++) {
		rc = peer_ins_ann(&addr[i], module[i], htl);
		if (0 != rc) {
			LOGS(L_PEER,L_MINOR,("peer_ins_ann(%s/%s) failed (%s)\n",
				sock_ntoa(&addr[i]),
				crypto_module_name(module[i]),
				strerror(errno)));
		}
	}

	return 0;
}

/*
 *	peer_req_host()
 *	Request a key with a chunk of data from a specific hostaddress.
 *	If that host is not yet connected or if the queue is full,
 *	request the chunk from other peers instead, that have a
 * 	routing fingerprint above 'threshold' for this key.
 */
static int peer_req_host(sha1_digest_t *s, struct sockaddr_in *peeraddr,
	int htl)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	time_t t0;
	size_t i;
	int req = 0, rc = 0;
	FUN("peer_req_host");

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);

	for (i = 0; 0 == req && i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != peeraddr->sin_addr.s_addr ||
			p->conn.address.sin_port != peeraddr->sin_port) {
			/* not the node we want */
			continue;
		}
		q = &p->queue;
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		e = &q->entry[q->head];
		e->cmd = MSG_REQ;
		e->htl = htl;
		e->sha1 = *s;
		e->size = 0;
		QUEUE_NEXT(q);
		req++;
		LOGS(L_PEER,L_DEBUG,("REQ %s from %s with HTL:%d on queue #%d/%d\n",
			sha1_hexshort(s), p->conn.peeraddr, htl, (int)i, (int)q->head));
	}
	GLOBAL_UNLOCK();

	/* If the peer could not be asked for that key... */
	if (0 == req) {
		size_t req;

		/* minimum useful hops-to-live for a request is 2 */
		if (htl < 2) {
			errno = EHOSTUNREACH;
			return -1;
		}

		rc = peer_req_key(s, 2, htl, &req, 0);
		if (0 == rc) {
			LOGS(L_PEER,L_DEBUG,("requested key %s from %u node(s) HTL:%d\n",
				sha1_hexshort(s), (unsigned)req, htl));
		} else {
			LOGS(L_PEER,L_DEBUG,("could not request key %s (%s)\n",
				sha1_hexshort(s), strerror(errno)));
		}
	}

	return rc;
}


/*
 *	peer_ins_host()
 *	Insert a key with a chunk of data for a specific hostaddress.
 *	If that host is not yet connected, publish it's hostaddress and
 *	send the block to other peers instead.
 */
static int peer_ins_host(sha1_digest_t *s, struct sockaddr_in *peeraddr,
	int htl, size_t size, uint8_t *buff)
{
	peer_node_t *p;
	msg_queue_t *q;
	msg_entry_t *e;
	time_t t0;
	size_t i;
	int ins = 0, rc = 0;
	FUN("peer_ins_host");

	if (htl > g_conf->maxhtl) {
		htl = g_conf->maxhtl;
		LOGS(L_PEER,L_DEBUG,("HTL clipped to %d!\n", htl));
	}

	if (0 != GLOBAL_LOCK())
		return -1;

	t0 = time(NULL);

	for (i = 0; 0 == ins && i < g_peer->peercnt_out; i++) {
		if (NULL == (p = g_peer->peers_out[i])) {
			/* unused slot */
			continue;
		}
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			/* not connected */
			continue;
		}
		if (p->conn.address.sin_addr.s_addr != peeraddr->sin_addr.s_addr ||
			p->conn.address.sin_port != peeraddr->sin_port) {
			/* not the node we want */
			continue;
		}
		q = &p->queue;
		if (0 != QUEUE_FULL(q)) {
			/* outbound queue is full */
			continue;
		}

		e = &q->entry[q->head];
		e->cmd = MSG_INS;
		e->htl = htl;
		e->sha1 = *s;
		e->size  = size;
		if (size <= MSG_BUFFSIZE) {
			/* data fits in buffer */
			memcpy(e->data, buff, size);
		} else {
			/* fetch data from file just in time */
			memset(e->data, 0, MSG_BUFFSIZE);
		}
		QUEUE_NEXT(q);
		ins++;
		LOGS(L_PEER,L_DEBUG,("INS %s to %s with HTL:%d on queue #%d/%d\n",
			sha1_hexshort(s), p->conn.peeraddr, htl,
			(int)i, (int)q->head));
	}
	GLOBAL_UNLOCK();

	/* If the key could not be inserted to that peer... */
	if (0 == ins) {
		size_t advertized;

		/* minimum useful hops-to-live for an advertizement is 2 */
		if (htl < 2) {
			errno = EHOSTUNREACH;
			return -1;
		}
		rc = peer_adv_key(s, 2, htl, &advertized, 0);
		if (0 == rc) {
			LOGS(L_PEER,L_DEBUG,("advertized key %s to %u node(s) HTL:%d\n",
				sha1_hexshort(s), (unsigned)advertized, htl));
		} else {
			LOGS(L_PEER,L_DEBUG,("could not advertize key %s (%s)\n",
				sha1_hexshort(s), strerror(errno)));
		}
	}

	return rc;
}

static int peer_out_msg(peer_node_t *p, peer_msg_t *msg, size_t *psize, int ini)
{
	uint32_t iosize;
	uint8_t *plaintext = p->plaintext;
	uint8_t *ciphertext = p->ciphertext;
	size_t insize = 0, outsize = 0;
	int rc = 0;
	FUN("peer_out_msg");

	LOGS(L_PEER,L_DEBUG,("%s key %s, htl %d peer %s\n",
		msg_name(msg->cmd),
		sha1_hexshort(&msg->sha1), msg->htl, p->conn.peeraddr));

	/* add some slack for the msg2stream conversion */
	insize = msg->size + 64;
	if (0 != (rc = msg2stream(plaintext, msg, &insize))) {
		LOGS(L_PEER,L_ERROR,("msg2stream() call failed (%s)\n",
			strerror(errno)));
		return -1;
	}

	/* if this is a non-initial message, go an encrypt it */
	if (0 == ini) {
		outsize = TEMPSIZE;
		rc = encrypt_msg(p->crypto_module_out, p->crypto_ctx,
			ciphertext, plaintext, &outsize, insize);
		if (0 != rc) {
			LOGS(L_PEER,L_ERROR,("encrypt_msg(%u,...,%#x,%#x)" \
				" to peer %s failed (%s)\n",
				(unsigned)p->crypto_module_out,
				(unsigned)outsize, (unsigned)insize,
				p->conn.peeraddr, strerror(errno)));
			return -1;
		}
	} else {
		outsize = insize;
		memcpy(ciphertext, plaintext, insize);
	}

	iosize = (uint32_t)htonl(outsize);
	if (0 != (rc = sock_writeall(&p->conn, &iosize, 4))) {
		LOGS(L_PEER,L_ERROR,("sock_writeall(%#x) [iosize]" \
			" to peer %s failed (%s)\n",
			(unsigned)4, p->conn.peeraddr, strerror(errno)));
		return -1;
	}

	if (0 != (rc = sock_writeall(&p->conn, ciphertext, outsize))) {
		LOGS(L_PEER,L_ERROR,("sock_writeall(%#x) [ciphertext]" \
			" to peer %s failed (%s)\n",
			(unsigned)outsize,
			p->conn.peeraddr, strerror(errno)));
		return -1;
	}

	BWLIMIT_LOCK();
	g_peer->out.total += outsize + sizeof(uint32_t);
	g_peer->out.daily += outsize + sizeof(uint32_t);
	BWLIMIT_UNLOCK();

	LOGS(L_PEER,L_DEBUG,("sent %#x bytes to %s\n",
		(unsigned)(outsize + sizeof(uint32_t)),
		p->conn.peeraddr));

	*psize = outsize;
	return rc;
}

/*
 *	peer_out_setup()
 *	set up a new outbound connection
 */
static int peer_out_setup(peer_node_t *p)
{
	struct sockaddr_in myname;
	osd_socklen_t myname_len = sizeof(myname);
	uint8_t version[128];
	uint32_t major, minor, build;
	peer_msg_t *msg = NULL;
	void *rsa = NULL;
	uint8_t *initial = NULL;
	uint8_t *pubkey = NULL;
	size_t pubkey_size = 0;
	uint8_t *encoded = NULL;
	size_t encoded_size = 0;
	char *xml = NULL;
	char *name = NULL;
	size_t offs, size;
	size_t initial_size;
	int rc = 0;
	FUN("peer_out_setup");

	/* Compare socket name to our IP address */
	rc = getsockname(p->conn.socket,
		(struct sockaddr *)&myname, &myname_len);
	if (0 == rc) {
		if (g_conf->node.sin_addr.s_addr == (uint32_t)-1) {
			/* If we were unresolveable before, set this address */
			g_conf->node.sin_addr.s_addr = myname.sin_addr.s_addr;
			LOGS(L_PEER,L_MINOR,("addr: %s, node: %s - set addr\n",
				sock_ntoa(&myname), sock_ntoa(&g_conf->node)));
		}
	}
	rc = 0;

	p->plaintext = xcalloc(TEMPSIZE + 256, sizeof(uint8_t));
	p->ciphertext = xcalloc(TEMPSIZE + 256, sizeof(uint8_t));

	offs = 0;
	size = 1;
	if (0 != (rc = sock_readall(&p->conn, &version[offs], size))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [version.nsize]" \
			" from %s failed (%s)\n",
			(unsigned)size,
			p->conn.peeraddr, strerror(errno)));
		goto bailout;
	}
	size = version[offs++];
	if (0 == size) {
		LOGS(L_PEER,L_ERROR,("version.name is too short (%#x)" \
			" from %s\n",
			(unsigned)size, p->conn.peeraddr));
		goto bailout;
	}
	if (offs + size > sizeof(version)) {
		LOGS(L_PEER,L_ERROR,("version.name is too long (%#x)" \
			" from %s\n",
			(unsigned)size, p->conn.peeraddr));
		goto bailout;
	}
	if (0 != (rc = sock_readall(&p->conn, &version[offs], size))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [version.name]" \
			" from %s failed (%s)\n",
			(unsigned)size,
			p->conn.peeraddr, strerror(errno)));
		goto bailout;
	}
	offs += size;
	version[offs++] = '\0';
	size = 1;
	if (0 != (rc = sock_readall(&p->conn, &version[offs], size))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [version.vsize]" \
			" from %s failed (%s)\n",
			(unsigned)size,
			p->conn.peeraddr, strerror(errno)));
		goto bailout;
	}
	size = version[offs++];
	if (offs + size > sizeof(version)) {
		LOGS(L_PEER,L_ERROR,("version.size is too long (%#x)" \
			" from %s\n",
			(unsigned)size, p->conn.peeraddr));
		goto bailout;
	}
	if (3*4 != size) {
		LOGS(L_PEER,L_ERROR,("version.size is not 12 (%#x)" \
			" from %s\n",
			(unsigned)size, p->conn.peeraddr));
		goto bailout;
	}
	if (0 != (rc = sock_readall(&p->conn, &version[offs], size))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [version.vers]" \
			" from %s failed (%s)\n",
			(unsigned)size,
			p->conn.peeraddr, strerror(errno)));
		goto bailout;
	}

	BWLIMIT_LOCK();
	g_peer->in.total += size;
	g_peer->in.daily += size;
	BWLIMIT_UNLOCK();

	major = ntohl(*(uint32_t*)&version[offs+0]);
	minor = ntohl(*(uint32_t*)&version[offs+4]);
	build = ntohl(*(uint32_t*)&version[offs+8]);

	if (MKVERSION2(major,minor) < PEER_MINVERSION ||
		build < PEER_MINBUILD) {
		LOGS(L_PEER,L_ERROR,("outgoing to %s/%s cancelled\n" \
			" peer build is old (%s-%u.%u.%u-%u)\n",
			p->conn.peeraddr, crypto_module_name(p->crypto_module_out),
			version + 1, major,
			MINOR_LEVEL(minor), MINOR_STEP(minor), build));
		goto bailout;
	}

	p->major = major;
	p->minor = minor;
	p->build = build;

	/* set some random initial fingerprint */
	rnd_get(p->fingerprint, sizeof(p->fingerprint));

	msg = (peer_msg_t *)xcalloc(1, sizeof(peer_msg_t) + CHUNKSIZE + 1);

	/* set name(s) for this peer node */
	if (0 == peer_get_name(&p->conn.address, &name)) {
		peer_set_name(&p->conn.address, name);
		xfree(name);
	}

	if (0 != (rc = rsa_init(&rsa))) {
		LOGS(L_PEER,L_ERROR,("rsa_init() call failed (%s)\n",
			strerror(errno)));
		goto bailout;
	}

	/* 256 bytes for now */
	initial_size = 256;
	initial = (uint8_t *)xcalloc(initial_size, sizeof(uint8_t));
	/* get initial PRNG data */
	if (0 != rnd_get(initial, initial_size)) {
		LOGS(L_PEER,L_ERROR,("rnd_get(...,%#x) call failed (%s)\n",
			(unsigned)initial_size, strerror(errno)));
		goto bailout;
	}
	rsa_strip(rsa, initial, initial_size);

	/*
	 * Initialize crypto module; starting with build 347 we check
	 * the crypto module's quality in a 'diequick' test (src/crypt.c)
	 */
	rc = crypto_init(p->crypto_module_out, &p->crypto_ctx,
		initial, initial_size, 1);

	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("crypto_init(%u,%p,%p,%#x) failed (%s)\n",
			(unsigned)p->crypto_module_out, p->crypto_ctx,
			initial, (unsigned)initial_size,
			strerror(errno)));
		goto bailout;
	}
	/* allocate space for a public key sent by the peer */
	pubkey_size = rsa_set_public_key(NULL, NULL, 0);
	LOGS(L_PEER,L_DEBUG,("pubkey_size is %#x\n",
		(unsigned)pubkey_size));
	pubkey = (uint8_t *)xcalloc(pubkey_size, sizeof(uint8_t));

	/* receive the public key from the peer */
	for (offs = 0; offs < pubkey_size; /* */) {
		size = pubkey_size - offs > CHUNKSIZE ? CHUNKSIZE : pubkey_size - offs;
		if (0 != (rc = sock_readall(&p->conn, &pubkey[offs], size))) {
			LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [pubkey part #%u]" \
				" from %s failed (%s)\n",
				(unsigned)size,
				(unsigned)(offs / CHUNKSIZE),
				p->conn.peeraddr, strerror(errno)));
			goto bailout;
		}
		LOGS(L_PEER,L_DEBUG,("received pubkey part #%u @%#x size %#x\n",
			(unsigned)(offs / CHUNKSIZE), (unsigned)offs, (unsigned)size));
		BWLIMIT_LOCK();
		g_peer->in.total += size;
		g_peer->in.daily += size;
		BWLIMIT_UNLOCK();
		offs += size;
	}
	LOGS(L_PEER,L_DEBUG,("received peer %s's pubkey (%#x bytes)\n",
		p->conn.peeraddr, (unsigned)pubkey_size));
	if (0 != (rsa_set_public_key(rsa, pubkey, pubkey_size))) {
		LOGS(L_PEER,L_ERROR,("rsa_set_public_key() call failed (%s)\n",
			strerror(errno)));
		goto bailout;
	}

	encoded_size = rsa_encode(rsa, NULL, NULL, initial_size);
	LOGS(L_PEER,L_DEBUG,("encoded_size(...,%#x) is %#x\n",
		(unsigned)initial_size, (unsigned)encoded_size));

	encoded = (uint8_t *)xcalloc(encoded_size, sizeof(uint8_t));
	if (0 != (rc = rsa_encode(rsa, initial, encoded, initial_size))) {
		LOGS(L_PEER,L_ERROR,("rsa_encode(%p,%p,%p,%#x) call failed (%s)\n",
			rsa, initial, encoded,
			(unsigned)initial_size, strerror(errno)));
		goto bailout;
	}

	xfree(pubkey);
	pubkey_size = 0;

	/* create an initial message to tell the peer about our nodeaddr */
	size = pm_asprintf(&xml,
		"<?xml version=\"1.0\" standalone='yes'?>\n" \
		"<p2p>\n" \
		"\t<peer address='%s' hostname='%s' port='%d' />\n" \
		"\t<node type='%s' major='%s' minor='%s' build='%s' />\n" \
		"\t<store fingerprint='%s' size='%llx' current='%llx' />\n" \
		"\t<crypto module='%s' size='%x' encoded='%x' initial='%s' />\n" \
		"</p2p>\n",
		sock_ntoa(&g_conf->node), g_conf->nodename, g_conf->nodeport,
		PROGNAME,NODE_VER_MAJ, NODE_VER_MIN, NODE_BUILD,
		hexstr(g_store->fingerprint, 16),
		g_store->storesize, g_store->currsize,
		crypto_module_name(p->crypto_module_out),
		(unsigned)initial_size, (unsigned)encoded_size,
		hexstr(encoded, encoded_size));

	if (size >= CHUNKSIZE) {
		LOGS(L_PEER,L_DEBUG,("Houston! We've got a problem: initial %#x\n",
			(unsigned)size));
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("initial message size is %#x\n",
		(unsigned)size));

	memcpy(msg->data, xml, size);
	xfree(xml);

	xfree(initial);
	initial_size = 0;

	xfree(encoded);
	encoded_size = 0;

	msg->cmd = MSG_ADV;
	msg->htl = 1;
	msg->size = size;
	sha1(msg->data, size, &msg->sha1);

	LOGS(L_PEER,L_DEBUG,("outgoing to %s/%s\n",
		p->conn.peeraddr, crypto_module_name(p->crypto_module_out)));

	if (0 != (rc = peer_out_msg(p, msg, &size, 1))) {
		LOGS(L_PEER,L_ERROR,("initial peer_out_msg() call failed\n"));
		goto bailout;
	}

	LOGS(L_PEER,L_NORMAL,("connection setup to %s/%s succeeded\n",
		p->conn.peeraddr, crypto_module_name(p->crypto_module_out)));

	xfree(msg);
	return 0;

bailout:
	LOGS(L_PEER,L_ERROR,("connection setup to %s/%s failed\n",
		p->conn.peeraddr, crypto_module_name(p->crypto_module_out)));
	rsa_destroy(&rsa);
	xfree(msg);
	xfree(initial);
	xfree(pubkey);
	xfree(encoded);
	xfree(xml);
	xfree(name);
	return -1;
}

/*
 *	peer_out_child()
 *	Run a peer outgoing connection. This function is called as a child
 *	process from the thread function for outgoing connections.
 *	The connection was established and the parent continues to create
 *	new connections.
 */
static void peer_out_child(peer_node_t *p)
{
	peer_msg_t *msg = NULL;
	uint8_t *buff = NULL;
	char *name = NULL;
	msg_queue_t *q;
	msg_entry_t *e;
	request_t *r;
	size_t i, j;
	size_t size;
	time_t tmin;
	uint64_t msgcount;
	uint64_t daily;
	long addr1, addr2;
	int local = 0;
	int rc = 0;
	FUN("peer_out_child");

	p->start = time(NULL);

	LOGS(L_PEER,L_MINOR,("new outbound connection to %s\n",
		p->conn.peeraddr));

	if (0 != peer_out_setup(p)) {
		LOGS(L_PEER,L_ERROR,("peer_out_setup() failed - good bye\n"));
		peer_add_failure(&p->conn.address);
		rc = -1;
		goto bailout;
	}

	addr1 = ntohl(p->conn.address.sin_addr.s_addr);
	addr2 = ntohl(g_conf->node.sin_addr.s_addr);
	local = is_local_ip(addr1, addr2);

	buff = (uint8_t *)xcalloc(TEMPSIZE + 256, sizeof(uint8_t));
	msg = (peer_msg_t *)xcalloc(1, sizeof(peer_msg_t) + CHUNKSIZE + 1);

	/* forever */
	msgcount = 0;
	for (;;) {
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			LOGS(L_PEER,L_MINOR,("socket was shut down\n"));
			goto bailout;
		}

		/* throttle if this appears to be a cancer node */
		if (p->cancer > 0) {
			/* 5 * cancer seconds (5 to 50) */
			osd_sleep(5 * p->cancer);
		}

		if (0 != (rc = GLOBAL_LOCK()))
			goto bailout;

		q = &p->queue;
		/* nothing in the queue? */
		if (q->tail == q->head) {
			GLOBAL_UNLOCK();
			/* 50ms delay */
			osd_usleep(50000);
			continue;
		} 

		e = &q->entry[q->tail];
		/* check for a cancelled message */
		if (MSG_CAN == e->cmd) {
			q->tail = (q->tail + 1) % q->size;
			GLOBAL_UNLOCK();
			/* 1ms delay */
			osd_usleep(1000);
			continue;
		}

		msg->htl = e->htl;
		msg->cmd = e->cmd;
		msg->sha1 = e->sha1;
		msg->size = e->size;
		if (msg->size > 0) {
			if (msg->size <= MSG_BUFFSIZE) {
				/* copy data from queue entry buffer */
				memcpy(msg->data, e->data, msg->size);
				q->tail = (q->tail + 1) % q->size;
				GLOBAL_UNLOCK();
				LOGS(L_PEER,L_MINOR,("Message from buff:\n" \
				"cmd:%02x htl:%02x key:%s size:%#x\n",
				msg->cmd, msg->htl, sha1_hexstr(&msg->sha1),
				(unsigned)msg->size));
				rc = 0;
			} else if (msg->size <= CHUNKSIZE) {
				q->tail = (q->tail + 1) % q->size;
				GLOBAL_UNLOCK();
				/* fetch data just in time */
				rc = store_get(&msg->sha1, buff);
				memcpy(msg->data, buff, msg->size);
				LOGS(L_PEER,L_MINOR,("Message from store:\n" \
				"cmd:%02x htl:%02x key:%s size:%#x rc:%d\n",
				msg->cmd, msg->htl, sha1_hexstr(&msg->sha1),
				(unsigned)msg->size, rc));
			} else {
				q->tail = (q->tail + 1) % q->size;
				GLOBAL_UNLOCK();
				LOGS(L_PEER,L_ERROR,("Bad message in queue:\n" \
				"cmd:%02x htl:%02x key:%s size:%#x\n",
				msg->cmd, msg->htl, sha1_hexstr(&msg->sha1),
				(unsigned)msg->size));
				rc = -1;
			}
		} else {
			q->tail = (q->tail + 1) % q->size;
			GLOBAL_UNLOCK();
		}
		if (0 != rc) {
			/* fetching from store failed, 50ms delay */
			osd_usleep(50000);
			continue;
		}

		if (MSG_INS == msg->cmd && 0 == msg->htl) {
			LOGS(L_PEER,L_ERROR,("INS msg with HTL 0, stupid!\n" \
				"cmd:%02x htl:%02x key:%s\n",
				msg->cmd, msg->htl, sha1_hexshort(&msg->sha1)));
			continue;
		}

		if (0 != (rc = peer_out_msg(p, msg, &size, 0))) {
			LOGS(L_PEER,L_MINOR,("peer_out_msg(%s) #%llu failed (%s)\n",
				sha1_hexshort(&msg->sha1), (unsigned long long)msgcount, strerror(errno)));
			goto bailout;
		}

		/* accounting */
		msgcount++;
		switch (msg->cmd) {
		case MSG_ADV:
			LOGS(L_PEER,L_DEBUG,("ADV %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.adv_second[59] += size;
			p->stats.adv_total += size;
			p->stats.adv_msgs += 1;
			STATS_UNLOCK();
			break;

		case MSG_REQ:
			LOGS(L_PEER,L_DEBUG,("REQ %s from %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.req_second[59] += size;
			p->stats.req_total += size;
			p->stats.req_msgs += 1;
			STATS_UNLOCK();
			/* Enter this request into the history */
			if (0 != (rc = GLOBAL_LOCK()))
				goto bailout;
			tmin = (time_t)0x7fffffff;
			for (i = 0, j = 0; i < REQUEST_MAX; i++) {
				r = &g_peer->req[i];
				if (0 == memcmp(&msg->sha1, &r->sha1, SHA1SIZE))
					break;
				if (r->time < tmin) {
					tmin = r->time;
					j = i;
				}
			}
			/* overwrite oldest entry if we didn't find this hash */
			if (i >= REQUEST_MAX) {
				r = &g_peer->req[j];
				r->sha1 = msg->sha1;
			}
			r->time = time(NULL);
			GLOBAL_UNLOCK();
			break;

		case MSG_INS:
			LOGS(L_PEER,L_DEBUG,("INS %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.ins_second[59] += size;
			p->stats.ins_total += size;
			p->stats.ins_msgs += 1;
			STATS_UNLOCK();
			break;

		case MSG_FPR:
			LOGS(L_PEER,L_DEBUG,("FPR %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.fpr_second[59] += size;
			p->stats.fpr_total += size;
			p->stats.fpr_msgs += 1;
			STATS_UNLOCK();
			break;

		case MSG_ANN:
			LOGS(L_PEER,L_DEBUG,("ANN %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.ann_second[59] += size;
			p->stats.ann_total += size;
			p->stats.ann_msgs += 1;
			STATS_UNLOCK();
			break;

		case MSG_FEC:
			LOGS(L_PEER,L_DEBUG,("FEC %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.fec_second[59] += size;
			p->stats.fec_total += size;
			p->stats.fec_msgs += 1;
			STATS_UNLOCK();
			break;

		case MSG_BCM:
			LOGS(L_PEER,L_DEBUG,("BCM %s to %s with HTL:%d\n",
				sha1_hexshort(&msg->sha1), p->conn.peeraddr, msg->htl));
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.bcm_second[59] += size;
			p->stats.bcm_total += size;
			p->stats.bcm_msgs += 1;
			STATS_UNLOCK();
			break;
		}

		BWLIMIT_LOCK();
		daily = g_peer->in.daily + g_peer->out.daily;
		BWLIMIT_UNLOCK();
		if (g_conf->bwlimit_daily > 0 && daily >= g_conf->bwlimit_daily) {
			LOGS(L_PEER,L_ERROR,("reached daily bandwidth limit %s - exiting\n",
				round_KMG(g_conf->bwlimit_daily)));
			break;
		}
		osd_usleep(1000);
	}

bailout:
	xfree(msg);
	xfree(buff);
	xfree(name);
	LOGS(L_PEER,L_MINOR,("closing outbound connection to %s\n",
		p->conn.peeraddr));
	peer_del_out(p);
	osd_exit(0);
}

/*
 *	peer_out()
 *	Start a thread that looks for peer addresses and establishes outgoing
 *	connections to the newly found peer addr:port tuples.
 */
int peer_out(void)
{
	char forkinfo[MAXPATHLEN];
	struct sockaddr_in peer;
	size_t crypto_module;
	uint64_t daily;
	conn_t conn;
	peer_node_t *p;
	pid_t pid;
	int rc = 0;
	FUN("peer_out");

	switch (osd_fork2("peer_out", g_conf->niceness, -1)) {
	case -1:
		LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
			"peer_out", g_conf->niceness, strerror(errno)));
		rc = -1;
		break;
	case 0:
		g_peer_out_pid = getpid();
		/* capture signals */
		set_signal_handler(SIGHUP, peer_out_exit);
		set_signal_handler(SIGINT, peer_out_exit);
		set_signal_handler(SIGPIPE, peer_out_exit);
		set_signal_handler(SIGALRM, peer_out_exit);
		set_signal_handler(SIGTERM, peer_out_exit);
		/* some delay before we start connecting nodes... */
		osd_sleep(15);
		/* forever... */
		for (;;) {

			if ((pid_t)-1 == g_peer_out_pid ||
				(pid_t)0 == g_peer_out_pid) {
				LOGS(L_PEER,L_ERROR,("g_peer_out_pid is invalid (%d)\n",
					(int)g_peer_out_pid));
				break;
			}

			BWLIMIT_LOCK();
			daily = g_peer->in.daily + g_peer->out.daily;
			BWLIMIT_UNLOCK();
			if (g_conf->bwlimit_daily > 0 && daily >= g_conf->bwlimit_daily) {
				LOGS(L_PEER,L_MINOR,("reached daily bandwidth limit %s - sleeping\n",
					round_KMG(g_conf->bwlimit_daily)));
				osd_sleep(60);
				continue;
			}

			rc = peer_node_search(&peer, &crypto_module);
			if (0 != rc) {
				/*
				 * delay = 1s * (peer_out_count + 1):
				 * no connection: 1s per search
				 * 31 connections: 32s per search
				 */
				osd_sleep(g_peer->peercnt_out + 1);
				continue;
			}
			if (0 != peer_blacklist_check(&peer)) {
				LOGS(L_PEER,L_ERROR,("peer %s rejected\n",
					sock_ntoa(&peer)));
				continue;
			}

			rc = peer_connected(&peer);
			if (0 == rc) {
				LOGS(L_PEER,L_DEBUG,("already connected %s; sleeping %us\n",
					sock_ntoa(&peer), (unsigned)(g_peer->peercnt_out + 1)));
				osd_sleep(g_peer->peercnt_out + 1);
				continue;
			}

			rc = peer_connecting(&peer);
			if (0 == rc) {
				LOGS(L_PEER,L_DEBUG,("already connecting %s; sleeping %us\n",
					sock_ntoa(&peer), (unsigned)(g_peer->peercnt_out + 1)));
				osd_sleep(g_peer->peercnt_out + 1);
				continue;
			}

			/* otherwise enter this peer into the list of tries */
			peer_add_connecting(&peer);

			if (g_peer->peercnt_out >= PEERS_OUT_MAX) {
				/* shut the least useful outbound connection */
				rc = peer_shut_worst_outbound();
				/* continue if no outbound connection was shut down */
				if (0 != rc) {
					LOGS(L_PEER,L_MINOR,("too many outgoing connections; don't connect to %s\n",
						sock_ntoa(&peer)));
					osd_sleep(30);
					continue;
				}
				osd_sleep(5);
			}

			pm_snprintf(forkinfo, sizeof(forkinfo), "peer_out %s/%s",
				sock_ntoa(&peer), crypto_module_name(crypto_module));
			LOGS(L_PEER,L_DEBUG,("trying peer %s/%s\n",
				sock_ntoa(&peer), crypto_module_name(crypto_module)));

			switch (osd_fork2(forkinfo, 0, -1)) {
			case -1:
				LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
					forkinfo, 0, strerror(errno)));
				break;
			case 0:
				pid = getpid();
				/* capture signals */
				set_signal_handler(SIGHUP, peer_out_exit);
				set_signal_handler(SIGINT, peer_out_exit);
				set_signal_handler(SIGPIPE, peer_out_exit);
				set_signal_handler(SIGALRM, peer_out_exit);
				set_signal_handler(SIGTERM, peer_out_exit);
				memset(&conn, 0, sizeof(conn));
				rc = sock_outgoing(&conn,inet_ntoa(peer.sin_addr),
					ntohs(peer.sin_port), 1);
				if (0 != rc) {
					LOGS(L_PEER,L_MINOR,("sock_outgoing(%s:%d) failed (%s)\n",
						inet_ntoa(peer.sin_addr),
						ntohs(peer.sin_port),
						strerror(errno)));
					peer_add_failure(&peer);
					osd_exit(0);
				}
				if (NULL == (p = peer_add_out())) {
					LOGS(L_PEER,L_ERROR,("peer_add_out() failed (%s)\n",
						strerror(errno)));
					osd_exit(0);
				}
				/* copy connection info */
				p->conn = conn;
				p->crypto_module_in = crypto_module;
				p->crypto_module_out = crypto_module;

				LOGS(L_PEER,L_MINOR,("sk %d connected to %s/%s (%s)\n",
					p->conn.socket, sock_ntoa(&peer),
					crypto_module_name(crypto_module),
					p->conn.peeraddr));

				(void)peer_out_child(p);
				LOGS(L_PEER,L_ERROR,("return from child %s (didn't exit)\n",
					sock_ntoa(&peer)));
				peer_del_out(p);
				osd_exit(0);
				break;
			default:
				osd_sleep(5);
			}
		}
	}

	return rc;
}

static int peer_in_msg(peer_node_t *p, peer_msg_t *msg, size_t *psize, int ini)
{
	uint32_t iosize;
	uint8_t *plaintext = p->plaintext;
	uint8_t *ciphertext = p->ciphertext;
	size_t insize = 0, outsize = 0;
	int rc = 0;
	FUN("peer_in_msg");

	/* read a block size */
	LOGS(L_PEER,L_DEBUG,("expecting %#x bytes from peer %s\n",
		(unsigned)4, p->conn.peeraddr));
	if (0 != (rc = sock_readall(&p->conn, &iosize, 4))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [iosize]" \
			" from peer %s failed\n",
			(unsigned)sizeof(uint32_t), p->conn.peeraddr));
		return -1;
	}
	insize = (size_t)ntohl(iosize);
	if (insize < MIN_MSGSIZE || insize > TEMPSIZE) {
		LOGS(L_PEER,L_ERROR,("bogus iosize %#x bytes from peer %s\n",
			(unsigned)insize, p->conn.peeraddr));
		errno = EINVAL;
		return -1;
	}
	LOGS(L_PEER,L_DEBUG,("expecting %#x bytes from peer %s\n",
		(unsigned)insize, p->conn.peeraddr));

	/* read the block data */
	if (0 != (rc = sock_readall(&p->conn, ciphertext, insize))) {
		LOGS(L_PEER,L_ERROR,("sock_readall(%#x) [ciphertext]" \
			" from peer %s failed\n",
			(unsigned)insize, p->conn.peeraddr));
		return -1;
	}

	BWLIMIT_LOCK();
	g_peer->in.total += insize + 4;
	g_peer->in.daily += insize + 4;
	BWLIMIT_UNLOCK();

	LOGS(L_PEER,L_DEBUG,("received %#x bytes from peer %s\n",
		(unsigned)(insize + 4), p->conn.peeraddr));

	/* If this is a non-initial message, go and decrypt the block */
	if (0 == ini) {
		outsize = TEMPSIZE;
		rc = decrypt_msg(p->crypto_module_out, p->crypto_ctx,
			plaintext, ciphertext, &outsize, insize);
		if (0 != rc) {
			LOGS(L_PEER,L_ERROR,("decrypt_msg(%u,...,%#x,%#x)" \
				" from peer %s failed\n",
				(unsigned)p->crypto_module_out,
				(unsigned)outsize, (unsigned)insize,
				p->conn.peeraddr));
			return -1;
		}
	} else {
		outsize = insize;
		memcpy(plaintext, ciphertext, insize);
	}

	if (outsize <= (size_t)4 || outsize > (size_t)(MAX_MSGSIZE + 4)) {
		LOGS(L_PEER,L_ERROR,("bogus size %#x bytes decrypted from %s\n",
			(unsigned)outsize, p->conn.peeraddr));
		errno = EINVAL;
		return -1;
	}

	/* convert the streamed data back into a msg */
	if (0 != (rc = stream2msg(msg, plaintext, outsize))) {
		LOGS(L_PEER,L_ERROR,("stream2msg() call failed from %s (%s)\n",
			p->conn.peeraddr, strerror(errno)));
		return -1;
	}

	LOGS(L_PEER,L_DEBUG,("%s key %s, HTL:%d peer %s\n",
		msg_name(msg->cmd), sha1_hexshort(&msg->sha1),
		msg->htl, p->conn.peeraddr));

	*psize = insize;
	return rc;
}

/*
 *	peer_in_setup()
 *	Setup a new inbound connection
 */
static int peer_in_setup(peer_node_t *p, struct sockaddr_in *peeraddr)
{
	struct sockaddr_in myname;
	osd_socklen_t myname_len = sizeof(myname);
	sha1_digest_t s;
	uint8_t version[128];
	uint32_t major, minor, build;
	size_t offs, size;
	p2p_info_t *pi = NULL;
	peer_msg_t *msg = NULL;
	uint8_t *buff = NULL;
	void *rsa = NULL;
	uint8_t *pubkey = NULL;
	size_t pubkey_size = 0;
	uint8_t *decoded = NULL;
	XML_Parser par = NULL;
	size_t decoded_size = 0;
	struct hostent *host;
	int rc = 0;
	FUN("peer_in_setup");

	/* Compare socket name to our IP address */
	rc = getsockname(p->conn.socket,
		(struct sockaddr *)&myname, &myname_len);
	if (0 == rc) {
		if (g_conf->node.sin_addr.s_addr == (uint32_t)-1) {
			/* If we were unresolveable before, set this address */
			g_conf->node.sin_addr.s_addr = myname.sin_addr.s_addr;
			LOGS(L_PEER,L_MINOR,("addr: %s, node: %s - set addr\n",
				sock_ntoa(&myname), sock_ntoa(&g_conf->node)));
		}
	}
	rc = 0;

	p->plaintext = xcalloc(TEMPSIZE + 256, sizeof(uint8_t));
	p->ciphertext = xcalloc(TEMPSIZE + 256, sizeof(uint8_t));

	/* fill a buffer containing the version and build numbers */
	major = strtoul(NODE_VER_MAJ, NULL, 0);
	minor = strtoul(NODE_VER_MIN, NULL, 0) * 65536 +
		strtoul(strchr(NODE_VER_MIN, '.') + 1, NULL, 0);
	build = strtoul(NODE_BUILD, NULL, 0);

	offs = 0;
	version[offs++] = strlen("EntropyRSA");
	offs += snprintf(&version[offs], sizeof(version) - offs,
		"EntropyRSA");
	version[offs++] = 3 * 4;
	*(uint32_t*)&version[offs] = htonl(major);
	offs += 4;
	*(uint32_t*)&version[offs] = htonl(minor);
	offs += 4;
	*(uint32_t*)&version[offs] = htonl(build);
	offs += 4;

	size = offs;
	if (0 != (rc = sock_writeall(&p->conn, version, size))) {
		LOGS(L_PEER,L_ERROR,("sock_writeall(%#x) [version]" \
			" to peer %s failed (%s)\n",
			(unsigned)size, p->conn.peeraddr, strerror(errno)));
		goto bailout;
	}
	BWLIMIT_LOCK();
	g_peer->out.total += size;
	g_peer->out.daily += size;
	BWLIMIT_UNLOCK();

	if (0 != (rc = rsa_init(&rsa))) {
		LOGS(L_PEER,L_ERROR,("rsa_init() call failed (%s)\n",
			strerror(errno)));
		goto bailout;
	}
	pubkey_size = rsa_get_public_key(NULL, NULL, 0);
	LOGS(L_PEER,L_DEBUG,("pubkey_size is %#x\n",
		(unsigned)pubkey_size));
	pubkey = (uint8_t *)xcalloc(pubkey_size, sizeof(uint8_t));
	if (0 != (rc = rsa_get_public_key(rsa, pubkey, pubkey_size))) {
		LOGS(L_PEER,L_ERROR,("rsa_get_public_key(%p,%p,%#x) call failed (%s)\n",
			rsa, pubkey, (unsigned)pubkey_size,
			strerror(errno)));
		goto bailout;
	}
	LOGS(L_PEER,L_DEBUG,("got our own pubkey (%#x)\n%s\n",
		(unsigned)pubkey_size, hexdump(pubkey, pubkey_size)));

	/* send the public key to the peer */
	for (offs = 0; offs < pubkey_size; /* */) {
		size = pubkey_size - offs > CHUNKSIZE ? CHUNKSIZE : pubkey_size - offs;
		if (0 != (rc = sock_writeall(&p->conn, &pubkey[offs], size))) {
			LOGS(L_PEER,L_ERROR,("sock_writeall(%#x) [pubkey part #%u]" \
				" to peer %s failed (%s)\n",
				(unsigned)size,
				(unsigned)(offs / CHUNKSIZE),
				p->conn.peeraddr, strerror(errno)));
			goto bailout;
		}
		LOGS(L_PEER,L_DEBUG,("sent pubkey part #%d @%#x size %#x\n",
			(unsigned)(offs / CHUNKSIZE), (unsigned)offs, (unsigned)size));
		BWLIMIT_LOCK();
		g_peer->out.total += size;
		g_peer->out.daily += size;
		BWLIMIT_UNLOCK();
		offs += size;
	}
	LOGS(L_PEER,L_DEBUG,("sent our own pubkey to %s (%#x bytes)\n",
		p->conn.peeraddr, (unsigned)pubkey_size));

	xfree(pubkey);
	pubkey_size = 0;

	msg = (peer_msg_t *)xcalloc(1, sizeof(peer_msg_t) + CHUNKSIZE + 1);

	/* get the initial message */
	memset(msg, 0, sizeof(peer_msg_t) + CHUNKSIZE);
	if (0 != (rc = peer_in_msg(p, msg, &size, 1))) {
		LOGS(L_PEER,L_ERROR,("initial peer_in_msg() call failed!\n"));
		goto bailout;
	}

	if (msg->size >= CHUNKSIZE) {
		LOGS(L_PEER,L_ERROR,("inital message too long (%#x) from %s\n",
			(unsigned)msg->size, p->conn.peeraddr));
		msg->size = CHUNKSIZE;
	}

	if (msg->size < 40) {
		LOGS(L_PEER,L_ERROR,("inital message too short (%#x) from %s\n",
			(unsigned)size, p->conn.peeraddr));
		goto bailout;
	}
	size = msg->size;

	sha1(msg->data, size, &s);
	if (0 != memcmp(&s, &msg->sha1, SHA1SIZE)) {
		LOGS(L_PEER,L_ERROR,("bad key %s of initial message from %s\n",
			sha1_hexshort(&msg->sha1), p->conn.peeraddr));
		goto bailout;
	}

	LOGS(L_PEER,L_DEBUG,("initial msg from %s\n", p->conn.peeraddr));

	if (NULL == (par = XML_ParserCreate(NULL))) {
		LOGS(L_PEER,L_ERROR,("XML_ParserCreate() failed\n"));
		errno = ENOMEM;
		goto bailout;
	}
	pi = (p2p_info_t *)xcalloc(1, sizeof(p2p_info_t));
	XML_SetElementHandler(par,
		(XML_StartElementHandler)p2p_info_xml_start_element,
		(XML_EndElementHandler)p2p_info_xml_end_element);
	XML_SetCharacterDataHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_cdata);
	XML_SetDefaultHandler(par,
		(XML_CharacterDataHandler)p2p_info_xml_default);
	XML_SetUserData(par, (void *)pi);
	if (0 == XML_Parse(par, msg->data, size, 1)) {
		rc = XML_GetErrorCode(par);
		LOGS(L_PEER,L_ERROR,("XML_Parse returned %d (line %d, pos %d)\n%s\n",
			rc, XML_GetCurrentLineNumber(par),
			XML_GetCurrentColumnNumber(par),
			msg->data));
		goto bailout;
	}
	XML_ParserFree(par);
	par = NULL;

	if (0 == pi->peer.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <peer ...> tag in initial message from %s\n",
			p->conn.peeraddr));
		errno = EINVAL;
		goto bailout;
	}

	if (ntohs(pi->peer.addr.sin_port) < 1024) {
		LOGS(L_PEER,L_ERROR,("Peer port %d is not acceptable\n",
			ntohs(pi->peer.addr.sin_port)));
		errno = EINVAL;
		goto bailout;
	}

	if (0 == pi->store.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <store ...> tag in initial message from %s\n",
			p->conn.peeraddr));
		errno = EINVAL;
		goto bailout;
	}

	if (0 == pi->crypto.found_tag) {
		LOGS(L_PEER,L_ERROR,("no <crypto ...> tag in initial message from %s\n",
			p->conn.peeraddr));
		errno = EINVAL;
		goto bailout;
	}

	if (0 == pi->node.found_tag) {
		LOGS(L_PEER,L_ERROR,("No <node ...> tag in initial message from %s\n",
			p->conn.peeraddr));
		errno = EINVAL;
		goto bailout;
	}

	if (0 == pi->crypto.encoded) {
		LOGS(L_PEER,L_ERROR,("no encoded attribute in initial <crypto ...> tag\n"));
		errno = EINVAL;
		goto bailout;
	}

	if (0 == pi->crypto.size) {
		LOGS(L_PEER,L_ERROR,("no size attribute in initial <crypto ...> tag\n"));
		errno = EINVAL;
		goto bailout;
	}

	pm_snprintf(p->type, sizeof(p->type), "%s", pi->node.type);
	p->major = pi->node.major;
	p->minor = pi->node.minor;
	p->build = pi->node.build;

	if (pi->node.build > g_conf->latest_build) {
		g_conf->latest_major = pi->node.major;
		g_conf->latest_minor = pi->node.minor;
		g_conf->latest_build = pi->node.build;
	}

	if (MKVERSION2(p->major,p->minor) < PEER_MINVERSION ||
		p->build < PEER_MINBUILD) {
		struct sockaddr_in addr = p->conn.address;
		LOGS(L_PEER,L_ERROR,("peer %s runs old %s-%d.%d.%d-%d; kicking\n",
			p->conn.peeraddr,
			p->type,
			p->major, MINOR_LEVEL(p->minor), MINOR_STEP(p->minor), p->build));
		addr.sin_port = -1;
		peer_add_failure(&addr);
		goto bailout;
	}

	if (0 == pi->store.fingerprint_sum) {
		if (0 != rnd_get(p->fingerprint, sizeof(p->fingerprint))) {
			LOGS(L_PEER,L_ERROR,("rnd_get() dummy fingerprint failed (%s)\n",
				strerror(errno)));
			goto bailout;
		}
	} else {
		memcpy(p->fingerprint, pi->store.fingerprint,
			sizeof(p->fingerprint));
	}

	p->crypto_module_in = p->crypto_module_out =
		crypto_module_indx(pi->crypto.module);

	decoded_size = rsa_decode(rsa, NULL, NULL, pi->crypto.encoded);
	LOGS(L_PEER,L_DEBUG,("decoded_size is %#x, crypto.size is %#x\n",
		(unsigned)decoded_size, (unsigned)pi->crypto.encoded));

	if (pi->crypto.size > decoded_size) {
		struct sockaddr_in addr;
		LOGS(L_PEER,L_ERROR,("initial size (%#x) > decoded size (%#x)\n",
			(unsigned)pi->crypto.size, (unsigned)decoded_size));
		addr = p->conn.address;
		addr.sin_port = -1;
		peer_add_failure(&addr);
		goto bailout;
	}

	/* allocate buffer for decoded session key (initial) */
	decoded = (uint8_t *)xcalloc(decoded_size, sizeof(uint8_t));
	rc = rsa_decode(rsa, pi->crypto.initial, decoded, pi->crypto.encoded);
	if (0 != rc) {
		struct sockaddr_in addr;
		LOGS(L_PEER,L_ERROR,("rsa_decode(%p,%p,%p,%#x) call failed (%s)\n",
			rsa, pi->crypto.initial, decoded,
			(unsigned)pi->crypto.encoded,
			strerror(errno)));
		addr = p->conn.address;
		addr.sin_port = -1;
		peer_add_failure(&addr);
		goto bailout;
	}

	/*
	 * Initialize crypto module; starting with build 347 we check
	 * the crypto module's quality in a 'diequick' test (src/crypt.c)
	 */
	rc = crypto_init(p->crypto_module_out, &p->crypto_ctx,
		decoded, pi->crypto.size, 1);

	memset(decoded, 0, decoded_size);
	xfree(decoded);
	decoded_size = 0;

	rsa_destroy(&rsa);

	if (0 != rc) {
		LOGS(L_PEER,L_ERROR,("crypto module '%s' init failed (%d)\n",
			pi->crypto.module, rc));
		errno = EINVAL;
		rc = -1;
		goto bailout;
	}

	LOGS(L_PEER,L_MINOR,("incoming node %s<->%s\n" \
		" version %s-%d.%d.%d-%d\n" \
		" fingerprint [%s]\n" \
		" crypto module: %s\n" \
		" initial size: %#x\n",
		p->conn.peeraddr, sock_ntoa(&pi->peer.addr), p->type,
		p->major, MINOR_LEVEL(p->minor), MINOR_STEP(p->minor), p->build,
		hexstr(p->fingerprint, 16),
		pi->crypto.module, (unsigned)pi->crypto.size));

	/* Use the incoming connection's IP address */
	peeraddr->sin_addr.s_addr = p->conn.address.sin_addr.s_addr;
	/* and the XML specified port */
	peeraddr->sin_port = pi->peer.addr.sin_port;

	if (0 == strlen(pi->peer.hostname) ||
		0 == strcasecmp(pi->peer.hostname, "UNKNOWN") ||
		0 == strcasecmp(pi->peer.hostname, "INVALID") ||
		0 == strcasecmp(pi->peer.hostname, "WINDOWS")) {
		host = gethostbyaddr((const char *)&peeraddr->sin_addr,
			sizeof(peeraddr->sin_addr), AF_INET);
		if (host == NULL) {
			LOGS(L_PEER,L_ERROR,("gethostbyaddr(%s,%d,AF_INET) failed (%s)\n",
				inet_ntoa(peeraddr->sin_addr),
				(int)sizeof(peeraddr->sin_addr),
				hstrerror(h_errno)));
			/* set IP number as name of peers with this address */
			peer_set_name(&p->conn.address,
				inet_ntoa(p->conn.address.sin_addr));
		} else {
			/* set names of peers with this address */
			peer_set_name(&p->conn.address, host->h_name);
		}
	} else {
		/* set names of peers with this address */
		peer_set_name(&p->conn.address, pi->peer.hostname);
	}

	peer_set_fingerprint(peeraddr, p->fingerprint);
	peer_set_crypto_module_in(peeraddr, p->crypto_module_out);
	peer_set_node(peeraddr,
		pi->node.type, pi->node.major, pi->node.minor, pi->node.build);
	if (pi->store.size > 0)
		peer_set_size_current(peeraddr, pi->store.size, pi->store.current);

	LOGS(L_PEER,L_NORMAL,("connection setup from %s/%s<->%s succeeded\n",
		p->conn.peeraddr, pi->crypto.module, sock_ntoa(&pi->peer.addr)));

	xfree(msg);
	xfree(buff);
	xfree(pi);
	return 0;

bailout:
	LOGS(L_PEER,L_NORMAL,("connection setup from %s failed\n",
		p->conn.peeraddr));
	if (NULL != par) {
		XML_ParserFree(par);
		par = NULL;
	}
	rsa_destroy(&rsa);
	xfree(pubkey);
	xfree(decoded);
	xfree(msg);
	xfree(buff);
	xfree(pi);
	return -1;
}

/*
 *	peer_in_child()
 *	Run a peer incoming connection. This function is called as a child
 *	process from the listening socket for incoming connections.
 *	The connection is accepted and the parent continues listening when
 *	we come here.
 */
static void peer_in_child(conn_t *conn)
{
	size_t i, size;
	uint64_t daily;
	uint64_t msgcount;
	peer_node_t *p = NULL;
	peer_msg_t *msg = NULL;
	request_t *r;
	uint8_t *buff = NULL;
	uint8_t avg, rnd;
	struct sockaddr_in peeraddr;
	size_t fpr;
	size_t advertized, requested;
	int htl_in, htl_out, rc = 0;
	FUN("peer_in_child");

	if (0 != peer_blacklist_check(&conn->address)) {
		LOGS(L_PEER,L_ERROR,("rejected %s\n", conn->peeraddr));
		osd_exit(0);
	}

	if (g_peer->peercnt_in >= PEERS_IN_MAX) {
		rc = peer_shut_worst_inbound();
		if (0 != rc) {
			LOGS(L_PEER,L_MINOR,("too many incoming connections; closing %s\n",
				conn->peeraddr));
			osd_exit(0);
		}
	}

	BWLIMIT_LOCK();
	daily = g_peer->in.daily + g_peer->out.daily;
	BWLIMIT_UNLOCK();
	if (g_conf->bwlimit_daily > 0 && daily >= g_conf->bwlimit_daily) {
		LOGS(L_PEER,L_MINOR,("reached daily bandwidth limit %s - exiting\n",
			round_KMG(g_conf->bwlimit_daily)));
		osd_exit(0);
	}

	/* give the shutdown process some time to make room in peer_in[] */
	osd_sleep(5);

	LOGS(L_PEER,L_MINOR,("adding incoming from %s\n",
		conn->peeraddr));

	if (NULL == (p = peer_add_in())) {
		LOGS(L_PEER,L_ERROR,("peer_add_in() failed (%s)\n",
			strerror(errno)));
		osd_exit(0);
	}
	p->conn = *conn;
	p->start = time(NULL);

	LOGS(L_PEER,L_MINOR,("new inbound connection from %s\n",
		p->conn.peeraddr));

	if (0 != peer_in_setup(p, &peeraddr)) {
		struct sockaddr_in addr = conn->address;
		LOGS(L_PEER,L_ERROR,("peer_in_setup() failed - good bye\n"));
		addr.sin_port = -1;
		peer_add_failure(&addr);
		peer_del_in(p);
		osd_exit(0);
	}

	peer_ins_ann(&peeraddr, p->crypto_module_out, ANNOUNCE_HTL);

	buff = (uint8_t *)xcalloc(TEMPSIZE+256, sizeof(uint8_t));
	msg = (peer_msg_t *)xcalloc(1, sizeof(peer_msg_t) + CHUNKSIZE + 1);

	/* forever */
	msgcount = 0;
	for (;;) {
		if (-1 == p->conn.socket || CONN_CONNECTED != p->conn.status) {
			LOGS(L_PEER,L_MINOR,("socket was shut down\n"));
			goto bailout;
		}

		/* throttle if this appears to be a cancer node */
		if (p->cancer > 0) {
			/* 5 * cancer seconds (5 to 50) */
			osd_sleep(5 * p->cancer);
		}

		if (0 != (rc = peer_in_msg(p, msg, &size, 0))) {
			LOGS(L_PEER,L_ERROR,("peer_in_msg() #%llu failed (%s)\n",
				(unsigned long long)msgcount, strerror(errno)));
			goto bailout;
		}
		htl_in = MSG_HTL(msg);
		/* clip htl value to configured maxhtl */
		if (htl_in > g_conf->maxhtl)
			htl_in = g_conf->maxhtl;
		/* Compute a decay for htl_out depending on # of outgoing */
		htl_out = (int)(htl_in * g_peer->peercnt_out / PEERS_OUT_MAX - 1);
		/* See if htl_out is really below 2 now */
		if (htl_out < 2)
			htl_out = htl_in - 1;
		/* clip htl value to configured maxhtl */
		if (htl_out > g_conf->maxhtl)
			htl_out = g_conf->maxhtl;

		/* fingeprint route of this key */
		fpr = keyroute(&msg->sha1);
		if (0 != rnd_get(&rnd, sizeof(rnd))) {
			LOGS(L_PEER,L_ERROR,("rnd_get(%u) failed (%s)\n",
				(unsigned)sizeof(rnd), strerror(errno)));
			goto bailout;
		}

		/* accounting */
		msgcount++;
		switch (msg->cmd) {
		case MSG_ADV:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.adv_second[59] += size;
			p->stats.adv_total += size;
			p->stats.adv_msgs += 1;
			p->keycount.total += 1;
			p->keycount.adv_count[fpr] += 1;
			STATS_UNLOCK();

			/* check the global request history if we requested this key */
			if (0 != GLOBAL_LOCK())
				goto bailout;
			for (i = 0; i < REQUEST_MAX; i++) {
				r = &g_peer->req[i];
				if (0 == memcmp(&msg->sha1, &r->sha1, SHA1SIZE)) {
					LOGS(L_PEER,L_MINOR,("pending REQ from %s answered\n",
						datetime_str(r->time)));
					if (r->time + 2*60 > time(NULL)) {
						/* zero the request history entry after 2 minutes */
						memset(r, 0, sizeof(*r));
					}
					break;
				}
			}
			GLOBAL_UNLOCK();

			if (i >= REQUEST_MAX) {
				/*
			 	* Request only keys which match this node's fingerprint for
			 	* 'fpr' better than 'avg', which is scaled depending on the
			 	* store's current size (empty stores take almost everything).
			 	*/
				avg = g_store->avg * g_store->currsize / g_store->storesize;
				if (g_store->fingerprint[fpr] < avg && avg < rnd) {
					LOGS(L_PEER,L_DEBUG,("ADV ignore %s from %s (%02x<%02x)\n",
						sha1_hexshort(&msg->sha1),
						sock_ntoa(&peeraddr),
						g_store->fingerprint[fpr], avg));
					p->stats.adv_msgs_ign += 1;
					/* penalty based on routing distance */
					penalty_delay(fpr, MSG_ADV, PENALTY_IGNORE);
					break;
				}
			}

			rc = store_get(&msg->sha1, msg->data);
			if (0 != rc) {
				p->stats.adv_msgs_miss += 1;
				rc = peer_req_host(&msg->sha1, &peeraddr, htl_in);
				if (0 == rc) {
					LOGS(L_PEER,L_DEBUG,("ADV requested key %s from %s\n",
						sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
				} else {
					/* penalty based on routing distance */
					penalty_delay(fpr, MSG_ADV, PENALTY_NHANDLE);
				}
				break;
			}

			size = CHUNKSIZE - 1;
			while (size > 0 && 0 == msg->data[size])
				size--;
			msg->size = ++size;
			if (0 != peer_msg_valid(msg)) {
				LOGS(L_PEER,L_ERROR,("ADV found bad/old key %s in store\n",
					sha1_hexshort(&msg->sha1)));
				store_zap(&msg->sha1);
				rc = peer_req_host(&msg->sha1, &peeraddr, htl_out);
				if (0 == rc) {
					LOGS(L_PEER,L_DEBUG,("ADV requested key %s from %s\n",
						sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
					p->stats.adv_msgs_miss += 1;
				} else {
					/* penalty based on routing distance */
					penalty_delay(fpr, MSG_ADV, PENALTY_NHANDLE);
				}
				break;
			}

			p->stats.adv_msgs_hit += 1;
			LOGS(L_PEER,L_DEBUG,("ADV already have key %s from %s\n",
				sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));

			/* Cancel further inserts of this key to this peer */
			peer_ins_cancel(&msg->sha1, &peeraddr);

			/* Advertize key to some peers */
			rc = -1;
			while (htl_out > 1) {
				rc = peer_adv_key(&msg->sha1, 2, htl_out, &advertized, 0);
				if (0 == rc)
					break;
				if (EINVAL == errno) {
					LOGS(L_PEER,L_ERROR,("peer_adv_key() EINVAL - bail out\n"));
					goto bailout;
				}
				LOGS(L_PEER,L_DEBUG,("adv_key(%s) for %s failed (%s)\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					strerror(errno)));
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_ADV, PENALTY_NFORWARD);
				htl_out--;
			}

			if (0 == rc) {
				p->stats.adv_msgs_fwd += advertized;
				LOGS(L_PEER,L_DEBUG,("adv_key(%s) for %s fwd %d times\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					(int)advertized));
				penalty_delay(fpr, MSG_ADV, PENALTY_SERVED);
			}
			break;

		case MSG_REQ:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.req_second[59] += size;
			p->stats.req_total += size;
			p->stats.req_msgs += 1;
			p->keycount.total += 1;
			p->keycount.req_count[fpr] += 1;
			STATS_UNLOCK();

			avg = g_store->avg;

			rc = store_get(&msg->sha1, msg->data);
			if (0 == rc) {
				size = CHUNKSIZE - 1;
				while (size > 0 && 0 == msg->data[size])
					size--;
				msg->size = ++size;
				if (0 != peer_msg_valid(msg)) {
					LOGS(L_PEER,L_ERROR,("REQ found bad/old key %s in store\n",
						sha1_hexshort(&msg->sha1)));
					store_zap(&msg->sha1);
					break;
				}
				if (size != msg->size) {
					/* find size again (trim trailing 00) */
					size = CHUNKSIZE - 1;
					while (size > 0 && 0 == msg->data[size])
						size--;
					msg->size = ++size;
				}
				/* valid key: reply if fingerprint is above avg */
				if (g_store->fingerprint[fpr] >= avg || avg >= rnd) {
					p->stats.req_msgs_hit += 1;
					LOGS(L_PEER,L_DEBUG,("REQ we have key %s for %s\n",
						sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
					while (htl_out > 0) {
						rc = peer_ins_host(&msg->sha1, &peeraddr, htl_out,
							msg->size, msg->data);
						if (0 == rc) {
							LOGS(L_PEER,L_DEBUG,("REQ inserted key %s for %s\n",
								sha1_hexshort(&msg->sha1),
								sock_ntoa(&peeraddr)));
							break;
						}
						/* penalty based on routing distance */
						penalty_delay(fpr, MSG_REQ, PENALTY_NHANDLE);
						htl_out--;
					}
					/* timed out or sent the key: we're done */
					break;
				}
				/* we're done with the key anyway */
				break;
			}

			/* HTL<2 (can't get a reply) end of chain: don't forward request */
			if (htl_out < 2) {
				p->stats.req_msgs_ign += 1;
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_REQ, PENALTY_NFORWARD);
				break;
			}

			/*
			 * React only on keys which match this node's fingerprint
			 * for 'fpr' better than 'avg', which is a lower limit for
			 * keys to keep. Let the random factor 'rnd' play a role
			 * in which keys to ignore, though.
			 */
			if (g_store->fingerprint[fpr] < avg && avg < rnd) {
				LOGS(L_PEER,L_DEBUG,("REQ ignore %s from %s (%02x<%02x)\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					g_store->fingerprint[fpr], avg));
				p->stats.req_msgs_ign += 1;
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_REQ, PENALTY_IGNORE);
				break;
			}
			p->stats.req_msgs_miss += 1;
			LOGS(L_PEER,L_DEBUG,("REQ we don't have key %s for %s\n",
				sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));

			/* request this key from some peers */
			rc = -1;
			while (htl_out > 1) {
				rc = peer_req_key(&msg->sha1, 2, htl_out, &requested, 0);
				if (0 == rc)
					break;
				if (EINVAL == errno) {
					LOGS(L_PEER,L_ERROR,("peer_req_key() EINVAL - bail out\n"));
					goto bailout;
				}
				LOGS(L_PEER,L_DEBUG,("req_key(%s) for %s failed (%s)\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					strerror(errno)));
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_REQ, PENALTY_NFORWARD);
				htl_out -= 1;
			}

			if (0 == rc) {
				p->stats.req_msgs_fwd += requested;
				LOGS(L_PEER,L_DEBUG,("req_key(%s) for %s fwd %d times\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					(int)requested));
				penalty_delay(fpr, MSG_REQ, PENALTY_SERVED);
			}
			break;

		case MSG_INS:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.ins_second[59] += size;
			p->stats.ins_total += size;
			p->stats.ins_msgs += 1;
			p->keycount.total += 1;
			p->keycount.ins_count[fpr] += 1;
			STATS_UNLOCK();

			/* check the global request history if we requested this key */
			if (0 != GLOBAL_LOCK())
				goto bailout;
			for (i = 0; i < REQUEST_MAX; i++) {
				r = &g_peer->req[i];
				if (0 == memcmp(&msg->sha1, &r->sha1, SHA1SIZE)) {
					LOGS(L_PEER,L_MINOR,("pending REQ from %s answered\n",
						datetime_str(r->time)));
					if (r->time + 2*60 > time(NULL)) {
						/* zero the request history entry after 2 minutes */
						memset(r, 0, sizeof(*r));
					}
					break;
				}
			}
			GLOBAL_UNLOCK();

			if (i >= REQUEST_MAX) {
				/*
			 	* Accept only keys which match this node's fingerprint for
			 	* 'fpr' better than 'avg', which is scaled depending on the
			 	* store's current size (empty stores take almost everything).
			 	*/
				avg = g_store->avg * g_store->currsize / g_store->storesize;
				/*
				 * now ignore the insert, if our fingeprint is below 'avg'
				 * and if the random factor 'rnd' doesn't change anything
				 * about our acceptance for this specific insert
				 */
				if (g_store->fingerprint[fpr] < avg && avg < rnd) {
					LOGS(L_PEER,L_DEBUG,("INS ignore %s from %s (%02x<%02x)\n",
						sha1_hexshort(&msg->sha1),
						sock_ntoa(&peeraddr),
						g_store->fingerprint[fpr], avg));
					p->stats.ins_msgs_ign += 1;
					/* penalty based on routing distance */
					penalty_delay(fpr, MSG_INS, PENALTY_IGNORE);
					break;
				}
			}

			/* check if this is an invalid try of a HTL <= 0 insert */
			if (htl_in <= 0) {
				LOGS(L_PEER,L_ERROR,("INS invalid msg (HTL:0) from %s\n",
					sock_ntoa(&peeraddr)));
				goto bailout;
			}

			/* check if this is an invalid try of a length == 0 insert */
			if (msg->size <= 0) {
				LOGS(L_PEER,L_ERROR,("INS invalid msg (size:%#x) from %s\n",
					(unsigned)msg->size, sock_ntoa(&peeraddr)));
				goto bailout;
			}

			/* check if this is an invalid try of appending 00s to data */
			if (0x00 == msg->data[msg->size - 1]) {
				LOGS(L_PEER,L_ERROR,("INS invalid msg (pad:00) from %s\n",
					sock_ntoa(&peeraddr)));
				goto bailout;
			}

			/* Check the validity of this message */
			if (0 != (rc = peer_msg_valid(msg))) {

				LOGS(L_PEER,L_ERROR,("*** INS bad key %s HTL:%d from %s\n",
					sha1_hexshort(&msg->sha1), htl_in, sock_ntoa(&peeraddr)));
				p->stats.ins_msgs_bad += 1;

				/* Increment cancer up to 10 */
				if (p->cancer < 10)
					p->cancer += 1;

				LOGS(L_PEER,L_NORMAL,("set peer %s cancer to %d\n",
					sock_ntoa(&p->conn.address), p->cancer));
				/* set this value for outbound connections to that node, too */
				peer_set_cancer(&peeraddr, p->cancer);
				break;
			}

			rc = store_get(&msg->sha1, buff);
			if (0 == rc) {
				size = CHUNKSIZE - 1;
				while (size > 0 && 0 == buff[size])
					size--;
				++size;
				if (0 != memcmp(buff, msg->data, msg->size)) {
					sha1_digest_t s;
					LOGS(L_PEER,L_ERROR,("INS different key %s from %s\n" \
						" his size: %#x\n" \
						" my size:  %#x\n" \
						" his type: %02x %02x %02x %02x\n" \
						" my type:  %02x %02x %02x %02x\n",
						sha1_hexshort(&msg->sha1),
						sock_ntoa(&peeraddr),
						(unsigned)msg->size,
						(unsigned)size,
						msg->data[0], msg->data[1], msg->data[2], msg->data[3],
						buff[0], buff[1], buff[2], buff[3]));
					sha1(msg->data, CHUNKSIZE, &s);
					if (0 == memcmp(&s, &msg->sha1, SHA1SIZE)) {
						if (0 != (rc = store_put(&msg->sha1, msg->data))) {
							LOGS(L_PEER,L_ERROR,("INS error storing key %s from %s\n",
								sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
							break;
						}
						LOGS(L_PEER,L_DEBUG,("INS stored key %s from %s\n",
							sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
					} else {
						p->stats.ins_msgs_bad += 1;
						/* Increment the node cancer count up to 10 */
						p->cancer += 1;
						if (p->cancer > 10)
							p->cancer = 10;

						LOGS(L_PEER,L_NORMAL,("set peer %s cancer to %d\n",
							sock_ntoa(&p->conn.address), p->cancer));
						/* set this value for outbound connections to that node */
						peer_set_cancer(&peeraddr, p->cancer);
						break;
					}
				}
				p->stats.ins_msgs_hit += 1;
			} else {
				if (0 != (rc = store_put(&msg->sha1, msg->data))) {
					LOGS(L_PEER,L_ERROR,("INS error storing key %s from %s\n",
						sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
					break;
				}
				LOGS(L_PEER,L_DEBUG,("INS stored key %s from %s\n",
					sha1_hexshort(&msg->sha1), sock_ntoa(&peeraddr)));
				p->stats.ins_msgs_miss += 1;
			}

			/* Cancel outstanding adv for this key to this peer */
			rc = peer_adv_cancel(&msg->sha1, &peeraddr);
			if (0 != rc && EINVAL == errno) {
				LOGS(L_PEER,L_ERROR,("peer_adv_cancel() EINVAL\n"));
				goto bailout;
			}
			/* Cancel all outstanding req for this key */
			rc = peer_req_cancel(&msg->sha1);
			if (0 != rc && EINVAL == errno) {
				LOGS(L_PEER,L_ERROR,("peer_req_cancel() EINVAL\n"));
				goto bailout;
			}

			if (0 != GLOBAL_LOCK())
				goto bailout;
			for (i = 0; i < REQUEST_MAX; i++) {
				r = &g_peer->req[i];
				if (0 == memcmp(&msg->sha1, &r->sha1, SHA1SIZE)) {
					LOGS(L_PEER,L_MINOR,("pending REQ from %s answered\n",
						datetime_str(r->time)));
					/* zero the request history entry */
					memset(r, 0, sizeof(*r));
					break;
				}
			}
			GLOBAL_UNLOCK();

			if (p->cancer > 0) {
				p->cancer -= 1;
				LOGS(L_PEER,L_NORMAL,("set peer %s cancer to %d\n",
					sock_ntoa(&p->conn.address), p->cancer));
				/* set this value for outbound connections to that node */
				peer_set_cancer(&peeraddr, p->cancer);
			}

			/* HTL<2 (can't get a request) end of chain: don't forward key */
			if (htl_out < 2) {
				p->stats.ins_msgs_ign += 1;
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_INS, PENALTY_NFORWARD);
				break;
			}

			/* Advertize to some peers */
			rc = -1;
			while (htl_out > 1) {
				rc = peer_adv_key(&msg->sha1, 2, htl_out, &advertized, 0);
				if (0 == rc)
					break;
				if (EINVAL == errno) {
					LOGS(L_PEER,L_ERROR,("peer_adv_key() EINVAL - bail out\n"));
					goto bailout;
				}
				LOGS(L_PEER,L_DEBUG,("adv_key(%s) for %s failed (%s)\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					strerror(errno)));
				/* penalty based on routing distance */
				penalty_delay(fpr, MSG_INS, PENALTY_NFORWARD);
				htl_out--;
			}

			if (0 == rc) {
				p->stats.ins_msgs_fwd += advertized;
				LOGS(L_PEER,L_DEBUG,("adv_key(%s) for %s fwd %d times\n",
					sha1_hexshort(&msg->sha1),
					sock_ntoa(&peeraddr),
					(int)advertized));
				penalty_delay(fpr, MSG_INS, PENALTY_SERVED);
			}
			break;

		case MSG_FPR:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.fpr_second[59] += size;
			p->stats.fpr_total += size;
			p->stats.fpr_msgs += 1;
			p->keycount.total += 1;
			p->keycount.fpr_count[fpr] += 1;
			STATS_UNLOCK();

			/* check if this is truely a fingerprint insert */
			if (htl_in > 0) {
				LOGS(L_PEER,L_ERROR,("FPR invalid (HTL:%d > 0) from %s\n",
					htl_in, sock_ntoa(&peeraddr)));
				goto bailout;
			}
			rc = peer_got_fpr(&peeraddr, msg);
			if (0 != rc) {
				LOGS(L_PEER,L_ERROR,("FPR invalid fingerprint from %s\n",
					sock_ntoa(&peeraddr)));
				goto bailout;
			}
			LOGS(L_PEER,L_DEBUG,("FPR got fingerprint from %s\n",
				sock_ntoa(&peeraddr)));
			break;

		case MSG_ANN:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.ann_second[59] += size;
			p->stats.ann_total += size;
			p->stats.ann_msgs += 1;
			p->keycount.total += 1;
			p->keycount.ann_count[fpr] += 1;
			STATS_UNLOCK();

			rc = peer_got_ann(&peeraddr, msg, htl_out);
			if (0 != rc) {
				LOGS(L_PEER,L_ERROR,("ANN invalid announcement from %s\n",
					sock_ntoa(&peeraddr)));
				goto bailout;
			}
			LOGS(L_PEER,L_DEBUG,("ANN got announcement from %s\n",
				sock_ntoa(&peeraddr)));
			break;

		case MSG_FEC:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.fec_second[59] += size;
			p->stats.fec_total += size;
			p->stats.fec_msgs += 1;
			p->keycount.total += 1;
			p->keycount.fec_count[fpr] += 1;
			STATS_UNLOCK();

			if (htl_out < 1)
				break;

			/* forward and/or handle the FEC request */
			rc = peer_got_fec(&peeraddr, msg, htl_out);

			if (0 !=  STATS_LOCK())
				goto bailout;
			if (0 == rc) {
				LOGS(L_PEER,L_DEBUG,("FEC request for %s succeeded\n",
					sock_ntoa(&peeraddr)));
				p->stats.fec_msgs_miss += 1;
			} else {
				int e = errno;
				LOGS(L_PEER,L_DEBUG,("FEC request for %s failed (%s)\n",
					sock_ntoa(&peeraddr), strerror(errno)));
				if (EINVAL == e) {
					p->stats.fec_msgs_bad += 1;
				} else if (EEXIST == e) {
					p->stats.fec_msgs_hit += 1;
				} else if (EAGAIN == e) {
					p->stats.fec_msgs_ign += 1;
				}
			}
			STATS_UNLOCK();
			break;

		case MSG_BCM:
			if (0 != (rc = STATS_LOCK()))
				goto bailout;
			p->stats.bcm_second[59] += size;
			p->stats.bcm_total += size;
			p->stats.bcm_msgs += 1;
			p->keycount.total += 1;
			p->keycount.bcm_count[fpr] += 1;
			STATS_UNLOCK();

			/* forward and/or handle the broadcast request */
			rc = peer_got_broadcast(&peeraddr, msg, htl_out);

			if (0 !=  STATS_LOCK())
				goto bailout;
			if (0 == rc) {
				LOGS(L_PEER,L_DEBUG,("BCM request from %s succeeded\n",
					sock_ntoa(&peeraddr)));
				p->stats.fec_msgs_miss += 1;
			} else {
				LOGS(L_PEER,L_DEBUG,("BCM request from %s failed (%s)\n",
					sock_ntoa(&peeraddr), strerror(errno)));
			}
			STATS_UNLOCK();
			break;


		default:
			LOGS(L_PEER,L_ERROR,("*** bogus message cmd %#x, HTL:%d from %s\n",
				msg->cmd, msg->htl, sock_ntoa(&p->conn.address)));
			goto bailout;
		}

		BWLIMIT_LOCK();
		daily = g_peer->in.daily + g_peer->out.daily;
		BWLIMIT_UNLOCK();
		if (g_conf->bwlimit_daily > 0 && daily >= g_conf->bwlimit_daily) {
			LOGS(L_PEER,L_ERROR,("reached daily bandwidth limit %s - exiting\n",
				round_KMG(g_conf->bwlimit_daily)));
			break;
		}
	}

bailout:
	xfree(msg);
	xfree(buff);
	LOGS(L_PEER,L_MINOR,("closing inbound connection from %s\n",
		sock_ntoa(&conn->address)));
	peer_del_in(p);
	osd_exit(0);
}

int peer_in_accept(void)
{
	/* detecting a shutdown condition */
	if ((pid_t)-1 == g_peer_in_pid ||
		(pid_t)0 == g_peer_in_pid) {
		return -1;
	}

	/* after each accepted connection, sleep depending on peercnt_in:
	 * delay = 1s * (peercnt_in + 1):
	 * no connection: 1s per connection
	 * 31 connections: 32s per search
	 */
	osd_sleep(g_peer->peercnt_in + 1);
	return 0;
}

/*
 *	peer_in()
 *	Start a thread that waits for incoming peer connections.
 */
int peer_in(void)
{
	char *listener = NULL;
	int rc = 0, bindport;
	FUN("peer_in");

	/* default to nodeport if bindport isn't explicitly set */
	bindport = g_conf->bindport;
	if (0 == bindport)
		bindport = g_conf->nodeport;
	rc = pm_asprintf(&listener, "peer_in [%s:%d]",
		g_conf->bindaddr, bindport);
	if (-1 == rc) {
		LOGS(L_PEER,L_ERROR,("pm_asprintf() call failed (%s)\n",
			strerror(errno)));
		return -1;
	}
	rc = 0;

	switch (osd_fork2(listener, g_conf->niceness, -1)) {
	case -1:
		LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
			listener, g_conf->niceness, strerror(errno)));
		xfree(listener);
		return -1;
	case 0:
		g_peer_in_pid = getpid();
		/* capture signals */
		set_signal_handler(SIGHUP, peer_in_exit);
		set_signal_handler(SIGINT, peer_in_exit);
		set_signal_handler(SIGPIPE, peer_in_exit);
		set_signal_handler(SIGALRM, peer_in_exit);
		set_signal_handler(SIGTERM, peer_in_exit);
		rc = sock_incoming(g_conf->bindaddr, bindport, 1,
			peer_in_child, peer_in_accept, 0, listener);
		if (0 != rc) {
			LOGS(L_PEER,L_ERROR,("sock_incoming(%s:%d,...) call failed (%s)\n",
				g_conf->bindaddr, g_conf->bindport, strerror(errno)));
		}
		osd_exit(rc);
	}
	xfree(listener);
	return rc;
}

#if	TICKS_PER_SEC
/*
 *	peer_bwlimit()
 *	Start a thread that limits the incoming and outgoing bandwidth by
 *	calling sock_set_bwlimit() once per second.
 */
int peer_bwlimit(void)
{
	int tickcount = 0, seccount = 0, mincount = 0, stats = 0;
	struct timeval tv, to;
	int seconds, number, val;
	int rc = 0;
	bandwidth_t *in, *out;
	uint64_t adv, req, ins, fec, fpr, ann, bcm, lim, iavg, oavg;
	peer_node_t *p;
	size_t i, j;
	uint64_t daily;
	FUN("peer_bwlimit");

	switch (osd_fork2("peer_bwlimit", g_conf->niceness, -1)) {
	case -1:
		LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
			"peer_bwlimit", g_conf->niceness, strerror(errno)));
		rc = -1;
		break;
	case 0:
		g_peer_bwlimit_pid = getpid();
		/* capture signals */
		set_signal_handler(SIGHUP, peer_bwlimit_exit);
		set_signal_handler(SIGINT, peer_bwlimit_exit);
		set_signal_handler(SIGPIPE, peer_bwlimit_exit);
		set_signal_handler(SIGALRM, peer_bwlimit_exit);
		set_signal_handler(SIGTERM, peer_bwlimit_exit);
		LOGS(L_PEER,L_MINOR,("bandwidth limiter running at %d ticks/sec\n",
			TICKS_PER_SEC));
		/* get starting date/time */
		gettimeofday(&tv, NULL);
		/* forever */
		for (;;) {
			if ((pid_t)-1 == g_peer_bwlimit_pid ||
				(pid_t)0 == g_peer_bwlimit_pid) {
				LOGS(L_PEER,L_ERROR,("g_peer_bwlimit_pid is invalid (%d)\n",
					(int)g_peer_bwlimit_pid));
				break;
			}
			osd_usleep(1000000/TICKS_PER_SEC);
			tickcount++;

			LOGS(L_PEER,L_DEBUGX,("tick #%d\n", tickcount));
			if (0 != BWLIMIT_LOCK()) {
				LOGS(L_PEER,L_ERROR,("locking bwlimit failed (%s)\n",
					strerror(errno)));
				break;
			}
			in = &g_peer->in;
			out = &g_peer->out;

			/*
			 * completely unrelated:
			 * stirr the internal PRNG with some entropy bits
			 * derived from the currently used bandwidth
			 * in-out or out-in
			 */
			if (0 == (tickcount & 1))
				val = in->used[59] - out->used[59];
			else
				val = out->used[59] - in->used[59];
			rnd_entropy (val);

			daily = g_peer->in.daily + g_peer->out.daily;
			if (g_conf->bwlimit_daily > 0 && daily >= g_conf->bwlimit_daily) {
				/*
				 * Do not give away any bandwidth as long as we exceed
				 * the daily limit defined ing g_conf->bwlimit_daily
				 */
				in->limit = 0;
				out->limit = 0;
			} else {
				lim = in->limit + g_conf->bwlimit_in / TICKS_PER_SEC;
				if (lim > g_conf->bwlimit_in / TICKS_PER_SEC) {
					lim = g_conf->bwlimit_in / TICKS_PER_SEC;
				}
				in->limit = lim;
				lim = out->limit + g_conf->bwlimit_out / TICKS_PER_SEC;
				if (lim > g_conf->bwlimit_out / TICKS_PER_SEC) {
					lim = g_conf->bwlimit_out / TICKS_PER_SEC;
				}
				out->limit = lim;
			}
			if (tickcount < TICKS_PER_SEC) {
				BWLIMIT_UNLOCK();
				continue;
			}
	
			tickcount -= TICKS_PER_SEC;
			iavg = 0;
			oavg = 0;
			for (j = 0; j < 60; j++) {
				iavg += in->used[j];
				oavg += out->used[j];
				/* move down the 'used' entries by one second */
				if (j < 59) {
					in->used[j] = in->used[j+1];
					out->used[j] = out->used[j+1];
				}
			}
			/* zap the last entry (which is where used bandwidth is added) */
			in->used[59] = 0;
			out->used[59] = 0;
			iavg /= 60;
			oavg /= 60;

			in->average = iavg;
			if (iavg >= in->peak) {
				/* if the average is over peak, set new peak */
				in->peak = iavg;
			} else {
				/* else decay the peak over time */
				in->peak = (uint32_t)((uint64_t)in->peak * 15 / 16);
			}
			/* weigh the real peak to the configured 50:50 */
			in->estimated = (uint32_t)
				(((uint64_t)in->peak + g_conf->bwlimit_in) / 2);
			/* no negative values: enforce a low available bandwidth */
			in->available = (in->estimated > in->average + 100) ?
				in->estimated - in->average : 100;

			out->average = oavg;
			if (oavg >= out->peak) {
				/* if the average is over peak, set new peak */
				out->peak = oavg;
			} else {
				/* else decay peak over time */
				out->peak = (uint32_t)((uint64_t)out->peak * 15 / 16);
			}
			/* weigh the real peak to the configured 50:50 */
			out->estimated = (uint32_t)
				(((uint64_t)out->peak + g_conf->bwlimit_out) / 2);
			out->available = (out->estimated > out->average + 20) ?
				out->estimated - out->average : 20;
			BWLIMIT_UNLOCK();

			if (++stats >= 10) {
				LOGS(L_PEER,L_MINOR,(
					"bandwidth estimation\n" \
					"measurings      in    out\n" \
					"-------------------------\n" \
					"last second %6u %6u\n" \
					"average/min %6u %6u\n" \
					"peak        %6u %6u\n" \
					"configured  %6u %6u\n" \
					"estimated   %6u %6u\n" \
					"-------------------------\n" \
					"available   %6u %6u\n",
					in->used[58], out->used[58],
					in->average, out->average,
					in->peak, out->peak,
					(unsigned)g_conf->bwlimit_in,
					(unsigned)g_conf->bwlimit_out,
					in->estimated, out->estimated,
					in->available, out->available));
				stats = 0;
			}

			if (0 != STATS_LOCK())
				break;
			for (i = 0; i < g_peer->peercnt_in; i++) {
				if (NULL == (p = g_peer->peers_in[i])) {
					continue;
				}
				if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
					continue;		/* skip closed connection peer */
				}
				adv = req = ins = fec = fpr = ann = bcm = 0;
				for (j = 0; j < 60; j++) {
					adv += p->stats.adv_second[j];
					req += p->stats.req_second[j];
					ins += p->stats.ins_second[j];
					fec += p->stats.fec_second[j];
					fpr += p->stats.fpr_second[j];
					ann += p->stats.ann_second[j];
					bcm += p->stats.bcm_second[j];
					if (j < 59) {
						p->stats.adv_second[j] = p->stats.adv_second[j+1];
						p->stats.req_second[j] = p->stats.req_second[j+1];
						p->stats.ins_second[j] = p->stats.ins_second[j+1];
						p->stats.fec_second[j] = p->stats.fec_second[j+1];
						p->stats.fpr_second[j] = p->stats.fpr_second[j+1];
						p->stats.ann_second[j] = p->stats.ann_second[j+1];
						p->stats.bcm_second[j] = p->stats.bcm_second[j+1];
					}
				}
				p->stats.adv_minute[59] = adv;
				p->stats.req_minute[59] = req;
				p->stats.ins_minute[59] = ins;
				p->stats.fec_minute[59] = fec;
				p->stats.fpr_minute[59] = fpr;
				p->stats.ann_minute[59] = ann;
				p->stats.adv_second[59] = 0;
				p->stats.req_second[59] = 0;
				p->stats.ins_second[59] = 0;
				p->stats.fec_second[59] = 0;
				p->stats.fpr_second[59] = 0;
				p->stats.ann_second[59] = 0;
				p->stats.bcm_second[59] = 0;
			}
			for (i = 0; i < g_peer->peercnt_out; i++) {
				if (NULL == (p = g_peer->peers_out[i])) {
					continue;
				}
				if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
					continue;		/* skip closed connection peer */
				}
				adv = req = ins = fec = fpr = ann = bcm = 0;
				for (j = 0; j < 60; j++) {
					adv += p->stats.adv_second[j];
					req += p->stats.req_second[j];
					ins += p->stats.ins_second[j];
					fec += p->stats.fec_second[j];
					fpr += p->stats.fpr_second[j];
					ann += p->stats.ann_second[j];
					bcm += p->stats.bcm_second[j];
					if (j < 59) {
						p->stats.adv_second[j] = p->stats.adv_second[j+1];
						p->stats.req_second[j] = p->stats.req_second[j+1];
						p->stats.ins_second[j] = p->stats.ins_second[j+1];
						p->stats.fec_second[j] = p->stats.fec_second[j+1];
						p->stats.fpr_second[j] = p->stats.fpr_second[j+1];
						p->stats.ann_second[j] = p->stats.ann_second[j+1];
						p->stats.bcm_second[j] = p->stats.bcm_second[j+1];
					}
				}
				p->stats.adv_minute[59] = adv;
				p->stats.req_minute[59] = req;
				p->stats.ins_minute[59] = ins;
				p->stats.fec_minute[59] = fec;
				p->stats.fpr_minute[59] = fpr;
				p->stats.ann_minute[59] = ann;
				p->stats.adv_second[59] = 0;
				p->stats.req_second[59] = 0;
				p->stats.ins_second[59] = 0;
				p->stats.fec_second[59] = 0;
				p->stats.fpr_second[59] = 0;
				p->stats.ann_second[59] = 0;
				p->stats.bcm_second[59] = 0;
			}

			seccount++;
			if (seccount >= 60) {
				/* update minute arrays */
				for (i = 0; i < g_peer->peercnt_in; i++) {
					if (NULL == (p = g_peer->peers_in[i])) {
						continue;
					}
					if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
						continue;		/* skip closed connection peer */
					}
					adv = req = ins = fec = fpr = ann = bcm = 0;
					for (j = 0; j < 60; j++) {
						adv += p->stats.adv_minute[j];
						req += p->stats.req_minute[j];
						ins += p->stats.ins_minute[j];
						fec += p->stats.fec_minute[j];
						fpr += p->stats.fpr_minute[j];
						ann += p->stats.ann_minute[j];
						bcm += p->stats.ann_minute[j];
						if (j < 59) {
							p->stats.adv_minute[j] = p->stats.adv_minute[j+1];
							p->stats.req_minute[j] = p->stats.req_minute[j+1];
							p->stats.ins_minute[j] = p->stats.ins_minute[j+1];
							p->stats.fec_minute[j] = p->stats.fec_minute[j+1];
							p->stats.fpr_minute[j] = p->stats.fpr_minute[j+1];
							p->stats.ann_minute[j] = p->stats.ann_minute[j+1];
							p->stats.bcm_minute[j] = p->stats.bcm_minute[j+1];
						}
					}
					p->stats.adv_hour[23] = adv;
					p->stats.req_hour[23] = req;
					p->stats.ins_hour[23] = ins;
					p->stats.fec_hour[23] = fec;
					p->stats.fpr_hour[23] = fpr;
					p->stats.ann_hour[23] = ann;
					p->stats.bcm_hour[23] = bcm;
				}
				for (i = 0; i < g_peer->peercnt_out; i++) {
					if (NULL == (p = g_peer->peers_out[i])) {
						continue;
					}
					if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
						continue;		/* skip closed connection peer */
					}
					adv = req = ins = fec = fpr = ann = bcm = 0;
					for (j = 0; j < 60; j++) {
						adv += p->stats.adv_minute[j];
						req += p->stats.req_minute[j];
						ins += p->stats.ins_minute[j];
						fec += p->stats.fec_minute[j];
						fpr += p->stats.fpr_minute[j];
						ann += p->stats.ann_minute[j];
						bcm += p->stats.bcm_minute[j];
						if (j < 59) {
							p->stats.adv_minute[j] = p->stats.adv_minute[j+1];
							p->stats.req_minute[j] = p->stats.req_minute[j+1];
							p->stats.ins_minute[j] = p->stats.ins_minute[j+1];
							p->stats.fec_minute[j] = p->stats.fec_minute[j+1];
							p->stats.fpr_minute[j] = p->stats.fpr_minute[j+1];
							p->stats.ann_minute[j] = p->stats.ann_minute[j+1];
							p->stats.bcm_minute[j] = p->stats.bcm_minute[j+1];
						}
					}
					p->stats.adv_hour[23] = adv;
					p->stats.req_hour[23] = req;
					p->stats.ins_hour[23] = ins;
					p->stats.fec_hour[23] = fec;
					p->stats.fpr_hour[23] = fpr;
					p->stats.ann_hour[23] = ann;
					p->stats.bcm_hour[23] = ann;
				}

				mincount++;
				if (mincount >= 60) {
					/* update hour arrays */
					for (i = 0; i < g_peer->peercnt_in; i++) {
						if (NULL == (p = g_peer->peers_in[i])) {
							continue;
						}
						if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
							continue;		/* skip closed connection peer */
						}
						adv = req = ins = fec = fpr = ann = bcm = 0;
						for (j = 0; j < 24; j++) {
							adv += p->stats.adv_hour[j];
							req += p->stats.req_hour[j];
							ins += p->stats.ins_hour[j];
							fec += p->stats.fec_hour[j];
							fpr += p->stats.fpr_hour[j];
							ann += p->stats.ann_hour[j];
							bcm += p->stats.bcm_hour[j];
							if (j < 23) {
								p->stats.adv_hour[j] = p->stats.adv_hour[j+1];
								p->stats.req_hour[j] = p->stats.req_hour[j+1];
								p->stats.ins_hour[j] = p->stats.ins_hour[j+1];
								p->stats.fec_hour[j] = p->stats.fec_hour[j+1];
								p->stats.fpr_hour[j] = p->stats.fpr_hour[j+1];
								p->stats.ann_hour[j] = p->stats.ann_hour[j+1];
								p->stats.bcm_hour[j] = p->stats.bcm_hour[j+1];
							}
						}
						p->stats.adv_24_hours = adv;
						p->stats.req_24_hours = req;
						p->stats.ins_24_hours = ins;
						p->stats.fec_24_hours = fec;
						p->stats.fpr_24_hours = fpr;
						p->stats.ann_24_hours = ann;
						p->stats.bcm_24_hours = bcm;
					}
					for (i = 0; i < g_peer->peercnt_out; i++) {
						if (NULL == (p = g_peer->peers_out[i])) {
							continue;
						}
						if (0 == p->pid || CONN_CONNECTED != p->conn.status) {
							continue;		/* skip closed connection peer */
						}
						adv = req = ins = fec = fpr = ann = 0;
						for (j = 0; j < 24; j++) {
							adv += p->stats.adv_hour[j];
							req += p->stats.req_hour[j];
							ins += p->stats.ins_hour[j];
							fec += p->stats.fec_hour[j];
							fpr += p->stats.fpr_hour[j];
							ann += p->stats.ann_hour[j];
							if (j < 23) {
								p->stats.adv_hour[j] = p->stats.adv_hour[j+1];
								p->stats.req_hour[j] = p->stats.req_hour[j+1];
								p->stats.ins_hour[j] = p->stats.ins_hour[j+1];
								p->stats.fec_hour[j] = p->stats.fec_hour[j+1];
								p->stats.fpr_hour[j] = p->stats.fpr_hour[j+1];
								p->stats.ann_hour[j] = p->stats.ann_hour[j+1];
							}
						}
						p->stats.adv_24_hours = adv;
						p->stats.req_24_hours = req;
						p->stats.ins_24_hours = ins;
						p->stats.fec_24_hours = fec;
						p->stats.fpr_24_hours = fpr;
						p->stats.ann_24_hours = ann;
					}
					mincount -= 60;
				}
				seccount -= 60;
			}
			STATS_UNLOCK();

			gettimeofday(&to, NULL);
			seconds = (int)(to.tv_sec - tv.tv_sec);
			if (seconds >= 24 * 60 * 60) {
				LOGS(L_PEER,L_NORMAL,("*** NEW DAY - RESET BANDWIDTH COUNTERS ***\n"));
				g_peer->in.daily = 0;
				g_peer->out.daily = 0;
				/* reset tv so that after 24 hours this code executes again */
				tv = to;
				/* reload seed nodes so that we have something to connect */
				peer_seednodes(&number);
			}
		}
		osd_exit(rc);
	}

	return rc;
}
#endif

static void peer_out_exit(int sig)
{
	pid_t pid = getpid();
	size_t i;
	peer_t *l_peer = g_peer;
	peer_node_t *p;
	int sk;
	FUN("peer_out_exit");

	signal(sig, SIG_DFL);
	LOGS(L_PEER,L_MINOR,("*** {%d} signal %s ***\n",
		(int)pid, signal_name(sig)));
	if (pid == g_peer_out_pid) {
		g_peer_out_pid = (pid_t)-1;
		return;
	}
	if (NULL == l_peer) {
		return;
	}
	for (i = 0; i < l_peer->peercnt_out; i++) {
		if (NULL == (p = l_peer->peers_out[i])) {
			continue;
		}
		if (pid != p->pid) {
			continue;
		}
		if (-1 == (sk = p->conn.socket)) {
			p->conn.status = CONN_CLOSED;
			continue;
		}
		p->conn.socket = -1;
		LOGS(L_PEER,L_NORMAL,("shutdown peer %s\n",
			sock_ntoa(&p->conn.address)));
		/* we cannot safely call shutdown() in a signal handler */ 
		osd_closesocket(sk);
		p->conn.status = CONN_CLOSED;
	}
}

static void peer_in_exit(int sig)
{
	pid_t pid = getpid();
	size_t i;
	peer_t *l_peer = g_peer;
	peer_node_t *p;
	int sk;
	FUN("peer_in_exit");

	signal(sig, SIG_DFL);
	LOGS(L_PEER,L_MINOR,("*** {%d} signal %s ***\n",
		(int)pid, signal_name(sig)));
	if (pid == g_peer_in_pid) {
		g_peer_in_pid = (pid_t)-1;
		return;
	}
	if (NULL == l_peer) {
		return;
	}
	for (i = 0; i < l_peer->peercnt_in; i++) {
		if (NULL == (p = l_peer->peers_in[i])) {
			continue;
		}
		if (pid != p->pid) {
			continue;
		}
		if (-1 == (sk = p->conn.socket)) {
			p->conn.status = CONN_CLOSED;
			continue;
		}
		p->conn.socket = -1;
		LOGS(L_PEER,L_NORMAL,("shutdown peer %s\n",
			sock_ntoa(&p->conn.address)));
		/* we cannot safely call shutdown() in a signal handler */ 
		osd_closesocket(sk);
		p->conn.status = CONN_CLOSED;
	}
}

static void peer_bwlimit_exit(int sig)
{
	pid_t pid = getpid();
	FUN("peer_bwlimit_exit");

	signal(sig, SIG_DFL);
	LOGS(L_PEER,L_MINOR,("*** {%d} signal %s ***\n",
		(int)getpid(), signal_name(sig)));
	GLOBAL_UNLOCK();
	BWLIMIT_UNLOCK();
	STATS_UNLOCK();
	FEC_UNLOCK();
	if (pid == g_peer_bwlimit_pid) {
		g_peer_bwlimit_pid = (pid_t)-1;
	}
}

static void peer_fec_worker_exit(int sig)
{
	pid_t pid = getpid();
	FUN("peer_fec_worker_exit");

	signal(sig, SIG_DFL);
	LOGS(L_PEER,L_MINOR,("*** {%d} signal %s ***\n",
		(int)pid, signal_name(sig)));
	GLOBAL_UNLOCK();
	BWLIMIT_UNLOCK();
	STATS_UNLOCK();
	FEC_UNLOCK();
	if (pid == g_peer_fec_worker_pid) {
		g_peer_fec_worker_pid = (pid_t)-1;
	}
}

static void peer_exit(int sig)
{
	pid_t pid = getpid();
	peer_t *l_peer = g_peer;
	FUN("peer_exit");

	signal(sig, SIG_DFL);
	LOGS(L_PEER,L_MINOR,("*** {%d} signal %s ***\n",
		(int)pid, signal_name(sig)));
	if (pid == g_peer_threads_pid) {
		g_peer_threads_pid = (pid_t)-1;
		osd_sem_destroy(&l_peer->sem_fec);
		osd_sem_destroy(&l_peer->sem_stats);
		osd_sem_destroy(&l_peer->sem_bwlimit);
		osd_sem_destroy(&l_peer->sem_global);
		LOGS(L_PEER,L_NORMAL,("shutdown g_peer sems {%d} %s\n",
			(int)pid, signal_name(sig)));
	}
}

/*
 * Create some noise on the line by requesting or advertizing a dummy key,
 * or sometimes even inserting some dummy data.
 */
static void peer_noise(void)
{
	sha1_digest_t s;
	uint8_t *buff = NULL;
	uint32_t seed, bsum;
	uint8_t dowhat;
	int htl;
	size_t i, size;
	FUN("peer_noise");

	/*
	 * re-seed the standard PRNG from our internal PRNG,
	 * then get a function 'dowhat' and dummy SHA1 hash.
	 * I used rand() to not leak too much info, though
	 * it might be possible to deduce the seed here...
	 * Should I use yet another PRNG, or our internal?
	 */
	if (0 != rnd_get(&seed, sizeof(seed))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("rnd_get() call failed (%s)\n", errmsg));
		die(1, "rnd_get() call failed (%s)\n", errmsg);
	}
	bsum = bitsum(seed);
	/* every now and then do nothing (too few/many bits, even number of bits) */
	if (bsum < 16-8 || bsum > 16+8 || 0 == (bsum & 1)) {
		LOGS(L_PEER,L_MINOR,("do nothing bsum(%#x)=%#x\n", seed, bsum));
		return;
	}
	srand(seed);
	for (i = 0; i < SHA1SIZE; i++)
		s.digest[i] = (rand() >> 20) & 0xff;
	dowhat = (rand() >> 7) & 0xff;
	htl = 1 + rand() % ((g_conf->maxhtl / 2) + 1);
	LOGS(L_PEER,L_MINOR,("do something bsum(%#x)=%#x, %#02x, HTL:%d\n",
		seed, bsum, dowhat, htl));

	if (dowhat < 0x80) {
		/* request a dummy key */
		LOGS(L_PEER,L_DEBUG,("REQ dummy key %s\n",
			sha1_hexshort(&s)));
		peer_req_key(&s, 2, htl, NULL, 0);
	} else if (dowhat > 0x80) {
		/* advertize a dummy key */
		LOGS(L_PEER,L_DEBUG,("ADV dummy key %s\n",
			sha1_hexshort(&s)));
		peer_adv_key(&s, 2, htl, NULL, 0);
	} else {
		/* dowhat = 0x80: create and advertize a dummy key */
		buff = xcalloc(CHUNKSIZE, sizeof(uint8_t));
		size = (rand() >> 5) & (CHUNKSIZE - 1);
		if (size < 2)
			size = 2;
		for (i = 0; i < size; i++)
			buff[i] = (uint8_t)(rand() >> 11);
		sha1(buff, CHUNKSIZE, &s);
		LOGS(L_PEER,L_DEBUGX,("ADV random key %s\n",
			sha1_hexshort(&s)));
		/* store the buffer */
		if (0 != store_put(&s, buff)) {
			LOGS(L_PEER,L_ERROR,("store_put(%s) call failed (%s)\n",
				sha1_hexshort(&s), strerror(errno)));
			xfree(buff);
			return;
		}
		xfree(buff);
		/* and advertize this key now */
		peer_adv_key(&s, 2, htl, NULL, 0);
		/* schedule a delete */
		if (0 != store_del(&s)) {
			LOGS(L_PEER,L_ERROR,("store_del(%s) call failed (%s)\n",
				sha1_hexshort(&s), strerror(errno)));
			return;
		}
	}
}

/*
 * Work down the next MSG_FEC request from the queue, if any.
 */
int peer_fec_worker(void)
{
	int rc = 0;
	FUN("peer_fec_worker");

	switch (osd_fork2("peer_fec_worker", g_conf->niceness, -1)) {
	case -1:
		LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
			"peer_fec_worker", g_conf->niceness, strerror(errno)));
		rc = -1;
		break;
	case 0:
		g_peer_fec_worker_pid = getpid();
		/* capture signals */
		set_signal_handler(SIGHUP, peer_fec_worker_exit);
		set_signal_handler(SIGINT, peer_fec_worker_exit);
		set_signal_handler(SIGPIPE, peer_fec_worker_exit);
		set_signal_handler(SIGALRM, peer_fec_worker_exit);
		set_signal_handler(SIGTERM, peer_fec_worker_exit);

		/* sleep a minute before we start polling the queue */
		osd_sleep(60);
		LOGS(L_PEER,L_MINOR,("MSG_FEC worker running, queue size: %u\n",
			(unsigned)g_peer->fec_queue.size));

		/* forever */
		for (;;) {
			size_t tail;
			sha1_digest_t s[1 + 12];
			fec_status_t st;
			fec_queue_t *f;
			fec_entry_t *e;
			int htl;

			/* if g_peer_fec_worker_pid becomes invalid, we're shutting down */
			if ((pid_t)-1 == g_peer_fec_worker_pid ||
				(pid_t)0 == g_peer_fec_worker_pid) {
				LOGS(L_PEER,L_ERROR,("g_peer_fec_worker_pid is invalid (%d)\n",
					(int)g_peer_fec_worker_pid));
				break;
			}

			if (0 != FEC_LOCK()) {
				LOGS(L_PEER,L_ERROR,("g_peer fec lock failed (%s)\n",
					strerror(errno)));
				break;
			}

			if (g_peer->peercnt_in < PEERS_IN_MAX/16 ||
				g_peer->peercnt_out < PEERS_OUT_MAX/16) {
				FEC_UNLOCK();
				LOGS(L_PEER,L_MINOR,("not enough connections (in:%u out:%u)\n",
					(unsigned)g_peer->peercnt_in,
					(unsigned)g_peer->peercnt_out));
				osd_sleep(30);
				continue;
			}

			f = &g_peer->fec_queue;

			if (f->size != g_conf->fec_queuesize) {
				LOGS(L_PEER,L_NORMAL,("changing queue size to %d\n",
					g_conf->fec_queuesize));
				/* take a new value from the config */
				f->size = g_conf->fec_queuesize;
				/* and reset head as well as tail to be safe */
				f->head = f->tail = 0;
			}

			if (0 == f->size) {
				/* silently sleep if queue size is zero */
				FEC_UNLOCK();
				osd_sleep(15);
				LOGS(L_PEER,L_DEBUG,("queue size 0: making noise\n"));
				peer_noise();
				continue;
			}

			if (f->tail == f->head) {
				FEC_UNLOCK();
				osd_sleep(15);
				LOGS(L_PEER,L_DEBUG,("empty queue: making noise\n"));
				peer_noise();
				continue;
			}

			/* get the tail entry */
			tail = f->tail;
			e = &f->entry[tail];
			memcpy(s, e->sha1, sizeof(s));
			st.sha1 = e->req;
			st.time = e->start;
			htl = e->htl;
			f->tail = (tail + 1) % f->size;
			if (0 != key_fec_status_get(&st)) {
				FEC_UNLOCK();
				LOGS(L_PEER,L_ERROR,("key_fec_status_get(%s) failed (%s)\n",
					sha1_hexshort(&st.sha1), strerror(errno)));
				osd_sleep(1);
				continue;
			}
			FEC_UNLOCK();

			LOGS(L_PEER,L_MINOR,("FEC %s w/ HTL:%d from #%#x dequeued (%s)\n",
				sha1_hexshort(&st.sha1),
				htl,
				(unsigned)tail,
				datetime_str(st.time)));

			rc = file_get_frag(s, htl);

			if (0 == rc) {
				st.status = FEC_DONE;
				LOGS(L_PEER,L_MINOR,("FEC %s w/ HTL:%d succeeded\n",
					sha1_hexshort(&st.sha1), htl));
			} else if (EINVAL == errno) {
				st.status = FEC_ERROR;
				LOGS(L_PEER,L_MINOR,("FEC %s is a bad requests (%s)\n",
					sha1_hexshort(&st.sha1), strerror(errno)));
			} else {
				st.status = FEC_UNUSED;
				st.retry += 1;
				LOGS(L_PEER,L_MINOR,("FEC %s HTL:%d try #%u failed (%s)\n",
					sha1_hexshort(&st.sha1), htl, (unsigned)st.retry,
					strerror(errno)));
			}
			if (0 != FEC_LOCK()) {
				LOGS(L_PEER,L_ERROR,("g_peer fec lock failed (%s)\n",
					strerror(errno)));
				break;
			}
			key_fec_status_set(&st);
			FEC_UNLOCK();

			osd_sleep(1);
		}
		osd_exit(0);
		break;
	}
	return rc;
}


int peer_threads(void)
{
	struct timeval t0, t1;
	int rc = 0;
	FUN("peer_threads");

	switch (osd_fork2("peer_threads", g_conf->niceness, -1)) {
	case -1:
		LOGS(L_PEER,L_ERROR,("osd_fork2('%s',%d) call failed (%s)\n",
			"peer_threads", g_conf->niceness, strerror(errno)));
		rc = -1;
		break;
	case 0:
		g_peer_threads_pid = getpid();
		/* capture signals */
		set_signal_handler(SIGHUP, peer_exit);
		set_signal_handler(SIGINT, peer_exit);
		set_signal_handler(SIGPIPE, peer_exit);
		set_signal_handler(SIGALRM, peer_exit);
		set_signal_handler(SIGTERM, peer_exit);

		/* get current time */
		gettimeofday(&t0, NULL);
		/* add 5 minutes */
		tv_add(&t1, &t0, (uint64_t)5 * 60 * 1000000);

		/* forever */
		for (;;) {
			/* sleep for some seconds */
			osd_sleep(15);

			/* if g_peer_threads_pid becomes invalid, we're shutting down */
			if ((pid_t)-1 == g_peer_threads_pid ||
				(pid_t)0 == g_peer_threads_pid) {
				LOGS(L_PEER,L_ERROR,("g_peer_threads_pid is invalid (%d)\n",
					(int)g_peer_threads_pid));
				break;
			}

			/* update configuration on the fly if entropy.conf changed */
			if (0 != (rc = conf_update())) {
				LOGS(L_PEER,L_ERROR,("conf_update() failed (%s)\n",
					strerror(errno)));
				break;
			}

			/* every five minutes call the peer node maintenance */
			gettimeofday(&t0, NULL);
			if (tv_cmp(&t1, &t0) <= 0) {
				LOGS(L_PEER,L_MINOR,("*** 5 MIN MAINTENANCE TIME ***\n"));

				/* announce all outgoing connections and kick hanging ones */
				LOGS(L_PEER,L_MINOR,("*** calling peer_node_maintenance()\n"));
				peer_node_maintenance();
				LOGS(L_PEER,L_MINOR,("*** back from peer_node_maintenance()\n"));

				/* purge the store if it is full */
				LOGS(L_PEER,L_MINOR,("*** calling store_purge()\n"));
				store_purge();
				LOGS(L_PEER,L_MINOR,("*** back from store_purge()\n"));

				gettimeofday(&t0, NULL);
				tv_add(&t1, &t0, (uint64_t)5 * 60 * 1000000);
			}
		}
		osd_exit(0);
		break;
	}

	return rc;
}

static int peer_seednodes(int *found)
{
	FILE *fp;
	char *line = NULL;
	struct sockaddr_in peeraddr;
	struct hostent *host;
	int number = 0, rc;
	FUN("peer_seednodes");

	rc = chdir(g_conf->progpath);
	if (-1 == rc) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("chdir(%s) failed (%s)\n",
			g_conf->progpath, errmsg));
		die(1, "chdir(%s) failed (%s)\n",
			g_conf->progpath, errmsg);
	}

	line = xcalloc(MAXPATHLEN, sizeof(char));

	fp = fopen(g_conf->seednodes, "r");
	if (NULL == fp) {
		FILE *fi;
		char *path;
		int lineno = 0;

		path = xcalloc(MAXPATHLEN, sizeof(char));
		pm_snprintf(path, MAXPATHLEN, "%s-dist", g_conf->seednodes);
		fi = fopen(path, "r");
		if (NULL == fi) {
			const char *errmsg = strerror(errno);
			LOGS(L_PEER,L_ERROR,("fopen(%s, 'r') failed (%s)\n",
				path, errmsg));
			die(1, "could not read seed nodes '%s' (%s)\n",
				path, errmsg);
		}
		fp = fopen(g_conf->seednodes, "w");
		if (NULL == fi || NULL == fp) {
			const char *errmsg = strerror(errno);
			LOGS(L_PEER,L_ERROR,("fopen(%s, 'w') failed (%s)\n",
				g_conf->seednodes, errmsg));
			die(1, "could not write seed nodes '%s' (%s)\n",
				g_conf->seednodes, errmsg);
		}
		while (!feof(fi)) {
			char *lineout = line;
			if (NULL == fgets(line, MAXPATHLEN, fi))
				break;
			if (lineno >= 2) {
				if (*lineout == '#')
					lineout++;
			}
			if (fputs(lineout, fp) < 0) {
				LOGS(L_PEER,L_ERROR,("could not write seed nodes '%s'\n",
					g_conf->seednodes));
				die(1, "could not write seed nodes '%s'\n",
					g_conf->seednodes);
			}
			lineno++;
		}
		xfree(path);
		fclose(fi);
		fclose(fp);
		fp = fopen(g_conf->seednodes, "r");
		if (NULL == fp) {
			const char *errmsg = strerror(errno);
			LOGS(L_PEER,L_ERROR,("could not open seed nodes '%s' (%s)\n",
				g_conf->seednodes, errmsg));
			die(1, "could not open seed nodes '%s' (%s)\n",
				g_conf->seednodes, errmsg);
		}
	}

	while (!feof(fp)) {
		char *str_addr, *str_port, *str_crypt;
		size_t crypto_module;

		if (NULL == fgets(line, MAXPATHLEN, fp))
			break;
		if (line[0] == '#' || line[0] == '\r' || line[0] == '\n')
			continue;
		str_addr = str_port = line;
		while (*str_port && *str_port != ':')
			str_port++;
		if (*str_port != ':') {
			info("\nWrong format '%s' in line number %d\n",
				g_conf->seednodes, number);
			continue;
		}
		*str_port++ = '\0';
		while (*str_port && isspace(*str_port))
			str_port++;

		str_crypt = str_port;
		while (*str_crypt && isdigit(*str_crypt))
			str_crypt++;
		if (*str_crypt != '\0') {
			while (*str_crypt && isspace(*str_crypt))
				*str_crypt++ = '\0';
			if ('\0' == *str_crypt ||
				'\r' == *str_crypt ||
				'\r' == *str_crypt) {
				str_crypt = g_conf->crypto_default;
			}
		}

		crypto_module = crypto_module_indx(str_crypt);
		/* check if this module is supported */
		if (-1 == crypto_module_supp(crypto_module)) {
			crypto_module = crypto_module_indx(g_conf->crypto_default);
		}

		LOGS(L_PEER,L_MINOR,("hostname:%s port:%s crypt:%s\n",
			str_addr, str_port, str_crypt));
		/* find out about the peer address */
		memset(&peeraddr, 0, sizeof(peeraddr));
		peeraddr.sin_family = AF_INET;
		peeraddr.sin_addr.s_addr = inet_addr(str_addr);
		peeraddr.sin_port = htons(strtoul(str_port, NULL, 0));
		if (peeraddr.sin_addr.s_addr == (uint32_t)-1) {
			/* inet_addr failed, probably a hostname */
			host = gethostbyname(str_addr);
			if (host == NULL) {
				LOGS(L_PEER,L_ERROR,("hostname '%s' does not resolve (%s)\n",
					str_addr, hstrerror(h_errno)));
				continue;
			}
			peeraddr.sin_addr.s_addr =
				((struct in_addr *) *host->h_addr_list)->s_addr;
		}

		number++;

		if (0 != peer_blacklist_check(&peeraddr)) {
			 LOGS(L_PEER,L_ERROR,("seed peer rejected\n"));
			 continue;
		}

		rc = peer_ins_ann(&peeraddr, crypto_module, 0);
		if (0 != rc) {
			LOGS(L_PEER,L_ERROR,("peer_ins_ann(%s,%s,0) failed (%s)\n",
				sock_ntoa(&peeraddr),
				crypto_module_name(crypto_module),
				strerror(errno)));
		}
	}
	fclose(fp);

	xfree(line);
	*found = number;
	return 0;
}

int peer(void)
{
	size_t size;
	struct hostent *host;
	int number = 0, rc = 0;
	FUN("peer");

	info("%s:%d, ", g_conf->nodename, g_conf->nodeport);
	/* find out about the peer address */
	memset(&g_conf->node, 0, sizeof(g_conf->node));
	g_conf->node.sin_family = AF_INET;
	g_conf->node.sin_addr.s_addr = inet_addr(g_conf->nodename);
	g_conf->node.sin_port = htons(g_conf->nodeport);
	if (g_conf->node.sin_addr.s_addr == (uint32_t)-1) {
		/* inet_addr failed, probably a hostname */
		host = gethostbyname(g_conf->nodename);
		if (host == NULL) {
			LOGS(L_PEER,L_ERROR,("hostname '%s' does not resolve (%s)\n",
				g_conf->nodename, hstrerror(h_errno)));
		} else {
			g_conf->node.sin_addr.s_addr =
				((struct in_addr *) *host->h_addr_list)->s_addr;
		}
	}

	size = sizeof(peer_t);
	g_peer = (peer_t *)scalloc(size, 1);
	if (0 != svalid(g_peer)) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("failed to shmalloc() peer (size %d) (%s)\n",
			(int)size, errmsg));
		die(1, "failed to shmalloc(%d) peer (%s)\n", (int)size, errmsg);
	}

	if (0 != (rc = osd_sem_init(&g_peer->sem_global, 1, 1))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_global, 1, 1, strerror(errno)));
		die(1, "osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_global, 1, 1, errmsg);
	}

	if (0 != (rc = osd_sem_init(&g_peer->sem_bwlimit, 1, 1))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_bwlimit, 1, 1, errmsg));
		die(1, "osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_bwlimit, 1, 1, errmsg);
	}

	if (0 != (rc = osd_sem_init(&g_peer->sem_stats, 1, 1))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_stats, 1, 1, errmsg));
		die(1, "osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_stats, 1, 1, errmsg);
	}

	if (0 != (rc = osd_sem_init(&g_peer->sem_fec, 1, 1))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_fec, 1, 1, errmsg));
		die(1, "osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_fec, 1, 1, errmsg);
	}

	if (0 != (rc = osd_sem_init(&g_peer->sem_bcm, 1, 1))) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_bcm, 1, 1, strerror(errno)));
		die(1, "osd_sem_init(%p,%d,%d) failed (%s)\n",
			&g_peer->sem_bcm, 1, 1, errmsg);
	}

	/* set the FEC queue size from the configuration */
	g_peer->fec_queue.size = g_conf->fec_queuesize;

	if (0 == g_conf->bwlimit) {
		/*
		 * just in case someone is configuring something stupid:
		 * if bwlimits are <= 0, set default 75K/25K
		 * if bwlimits are <= 100, assume (s)he meant K and multiply by 1024
		 */
		if (g_conf->bwlimit_in <= 0)
			g_conf->bwlimit_in = 75*1024;
		else if (g_conf->bwlimit_in < 100)
			g_conf->bwlimit_in *= 1024;

		if (g_conf->bwlimit_out <= 0)
			g_conf->bwlimit_out = 25*1024;
		else if (g_conf->bwlimit_out < 100)
			g_conf->bwlimit_out *= 1024;
	} else {
		/* divide bwlimit into 3/4th and 1/4th for in and out */
		g_conf->bwlimit_in = g_conf->bwlimit * 3 / 4;
		g_conf->bwlimit_out = g_conf->bwlimit * 1 / 4;
	}
	LOGS(L_PEER,L_NORMAL,("bandwidth limits in: %d, out: %d\n",
		(int)g_conf->bwlimit_in, (int)g_conf->bwlimit_out));

#if	TICKS_PER_SEC
	/* start the bandwidth limiter thread */
	if (0 != (rc = peer_bwlimit())) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("peer_bwlimit() call failed (%s)\n", errmsg));
		die(rc, "peer_bwlimit() failed (%s)", errmsg);
	}
#endif

	if (0 != (rc = peer_threads())) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("peer_threads() call failed (%s)\n",
			errmsg));
		die(1, "peer_threads() call failed (%s)\n", errmsg);
	}

	/* start the outgoing connections thread */
	if (0 != (rc = peer_out())) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("peer_out() call failed (%s)\n", errmsg));
		die(rc, "peer_out() failed (%s)", errmsg);
	}

	/* start the incoming connections thread */
	if (0 != (rc = peer_in())) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("peer_in() call failed (%s)\n", errmsg));
		die(rc, "peer_in() failed (%s)", errmsg);
	}

	/* start the MSG_FEC worker thread */
	if (0 != (rc = peer_fec_worker())) {
		const char *errmsg = strerror(errno);
		LOGS(L_PEER,L_ERROR,("peer_fec_worker() call failed (%s)\n",
			errmsg));
		die(rc, "peer_fec_worker() failed (%s)", errmsg);
	}

	peer_seednodes(&number);
	if (0 == number) {
		LOGS(L_PEER,L_ERROR,("No seed nodes found in '%s'\n",
			g_conf->seednodes));
		die(1, "No seed nodes found in '%s'", g_conf->seednodes);
	}
	info("%d seed nodes ", number);

	return rc;
}
